ladylilia.com/twitch-subs-api/app.py

import logging
import os

import flask
import flask_login
import flask_restful
import flask_restful.fields
import flask_restful.reqparse
import itsdangerous
import sqlalchemy
import sqlalchemy.engine

from db import db, RegularSub, GiftedSub


app = flask.Flask(__name__)
app.logger.setLevel(logging.INFO)
app.config.update(
    ERROR_404_HELP=False,
    SQLALCHEMY_TRACK_MODIFICATIONS=False,
    SQLALCHEMY_DATABASE_URI=os.getenv('SQLALCHEMY_DATABASE_URI'),
    SECRET_KEY=os.getenv('SECRET_KEY'))

if app.config.get('SQLALCHEMY_DATABASE_URI', '').startswith('sqlite://'):
    @sqlalchemy.event.listens_for(sqlalchemy.engine.Engine, 'connect')
    def set_sqlite_pragma(dbapi_connection, connection_record):
        dbapi_connection.execute('PRAGMA journal_mode=WAL')
        dbapi_connection.execute('PRAGMA synchronous=NORMAL')

db.init_app(app)
db.create_all(app=app)

login_manager = flask_login.LoginManager()
login_manager.init_app(app)

api = flask_restful.Api(app)


regular_sub_fields = {
    'id': flask_restful.fields.Integer(),
    'user': flask_restful.fields.String(),
    'rank': flask_restful.fields.Integer(),
    'time': flask_restful.fields.DateTime(dt_format='iso8601'),
    'created_at': flask_restful.fields.DateTime(dt_format='iso8601'),
    'updated_at': flask_restful.fields.DateTime(dt_format='iso8601'),
}

gifted_sub_fields = {
    'id': flask_restful.fields.Integer(),
    'giver': flask_restful.fields.String(),
    'receiver': flask_restful.fields.String(),
    'time': flask_restful.fields.DateTime(dt_format='iso8601'),
    'created_at': flask_restful.fields.DateTime(dt_format='iso8601'),
    'updated_at': flask_restful.fields.DateTime(dt_format='iso8601'),
}


regular_sub_parser = flask_restful.reqparse.RequestParser()
regular_sub_parser.add_argument('id', type=int)
regular_sub_parser.add_argument('user', type=str, required=True)
regular_sub_parser.add_argument('rank', type=int, required=True)
regular_sub_parser.add_argument('time', type=flask_restful.inputs.datetime_from_iso8601, required=True)

gifted_sub_parser = flask_restful.reqparse.RequestParser()
gifted_sub_parser.add_argument('id', type=int)
gifted_sub_parser.add_argument('giver', type=str, required=True)
gifted_sub_parser.add_argument('receiver', type=str, required=True)
gifted_sub_parser.add_argument('time', type=flask_restful.inputs.datetime_from_iso8601, required=True)

filter_parser = flask_restful.reqparse.RequestParser()
filter_parser.add_argument('filter', type=str)
filter_parser.add_argument('older_than', type=flask_restful.inputs.datetime_from_iso8601)
filter_parser.add_argument('newer_than', type=flask_restful.inputs.datetime_from_iso8601)
filter_parser.add_argument('sort_by', type=str)
filter_parser.add_argument('sort_order', type=str)
filter_parser.add_argument('page_number', type=int)
filter_parser.add_argument('page_size', type=int)


@login_manager.request_loader
def load_user(request):
    key = request.headers.get('X-Twitch-Subs-API-Key')
    if not key:
        return None
    s = itsdangerous.TimedJSONWebSignatureSerializer(app.config['SECRET_KEY'])
    try:
        user = flask_login.UserMixin()
        user.id = s.loads(key)
        return user
    except (itsdangerous.SignatureExpired, itsdangerous.BadSignature):
        return None


class RegularSubResource(flask_restful.Resource):
    @flask_restful.marshal_with(regular_sub_fields)
    def get(self, id):
        q = db.session.query(RegularSub).filter(RegularSub.id == id)
        regular_sub = q.first()
        if not regular_sub:
            flask_restful.abort(404, message='Regular sub {0} does not exist'.format(id))
        return regular_sub, 200

    @flask_login.login_required
    @flask_restful.marshal_with(regular_sub_fields)
    def put(self, id):
        args = regular_sub_parser.parse_args()
        now = sqlalchemy.func.now()
        q = db.session.query(RegularSub).filter(RegularSub.id == id)
        regular_sub = q.first()
        if not regular_sub:
            regular_sub = RegularSub(id=id, created_at=now)
        regular_sub.user = args['user']
        regular_sub.rank = args['rank']
        regular_sub.time = args['time']
        regular_sub.updated_at = now
        db.session.add(regular_sub)
        db.session.commit()
        return regular_sub, 200

    @flask_login.login_required
    def delete(self, id):
        q = db.session.query(RegularSub).filter(RegularSub.id == id)
        regular_sub = q.first()
        if not regular_sub:
            flask_restful.abort(404, message='Regular sub {0} does not exist'.format(id))
        db.session.delete(regular_sub)
        db.session.commit()
        return None, 204


class RegularSubsResource(flask_restful.Resource):
    @flask_restful.marshal_with(regular_sub_fields)
    def get(self):
        args = filter_parser.parse_args()
        q = db.session.query(RegularSub)
        if args['filter']:
            q = q.filter(RegularSub.user.ilike('%{}%'.format(args['filter'])))
        if args['older_than']:
            q = q.filter(RegularSub.time < args['older_than'])
        if args['newer_than']:
            q = q.filter(RegularSub.time > args['newer_than'])
        count = q.count()
        if args['sort_order'] == 'random':
            q = q.order_by(sqlalchemy.func.random())
        elif args['sort_by']:
            col = getattr(RegularSub, args['sort_by'], None)
            if col:
                if args['sort_order']:
                    order_by = getattr(col, args['sort_order'], None)
                    if order_by:
                        q = q.order_by(order_by())
                else:
                    q = q.order_by(col)
        if args['page_size']:
            q = q.limit(args['page_size'])
        if args['page_number'] and args['page_size']:
            q = q.offset(args['page_number'] * args['page_size'])
        regular_subs = q.all()
        return regular_subs, 200, {'X-Total-Count': count}

    @flask_login.login_required
    @flask_restful.marshal_with(regular_sub_fields)
    def post(self):
        args = regular_sub_parser.parse_args()
        if not args['user']:
            flask_restful.abort(400, message='Missing required parameter user')
        if not args['rank']:
            flask_restful.abort(400, message='Missing required parameter rank')
        if not args['time']:
            flask_restful.abort(400, message='Missing required parameter time')
        now = sqlalchemy.func.now()
        q = db.session.query(RegularSub).filter((RegularSub.user == args['user']) &\
                (RegularSub.rank == args['rank']) &\
                (sqlalchemy.func.DATE(RegularSub.time) == args['time'].date()))
        regular_sub = q.first()
        if not regular_sub:
            regular_sub = RegularSub(created_at=now)
        regular_sub.user = args['user']
        regular_sub.rank = args['rank']
        regular_sub.time = args['time']
        regular_sub.updated_at = now
        db.session.add(regular_sub)
        db.session.commit()
        url = api.url_for(RegularSubResource, id=regular_sub.id, _external=True, _scheme='https')
        return regular_sub, 201, {'Location': url}


class GiftedSubResource(flask_restful.Resource):
    @flask_restful.marshal_with(gifted_sub_fields)
    def get(self, id):
        q = db.session.query(GiftedSub).filter(GiftedSub.id == id)
        gifted_sub = q.first()
        if not gifted_sub:
            flask_restful.abort(404, message='Gifted sub {0} does not exist'.format(id))
        return gifted_sub, 200

    @flask_login.login_required
    @flask_restful.marshal_with(gifted_sub_fields)
    def put(self, id):
        args = gifted_sub_parser.parse_args()
        now = sqlalchemy.func.now()
        q = db.session.query(GiftedSub).filter(GiftedSub.id == id)
        gifted_sub = q.first()
        if not gifted_sub:
            gifted_sub = GiftedSub(id=id, created_at=now)
        gifted_sub.giver = args['giver']
        gifted_sub.receiver = args['receiver']
        gifted_sub.time = args['time']
        gifted_sub.updated_at = now
        db.session.add(gifted_sub)
        db.session.commit()
        return gifted_sub, 200

    @flask_login.login_required
    def delete(self, id):
        q = db.session.query(GiftedSub).filter(GiftedSub.id == id)
        gifted_sub = q.first()
        if not gifted_sub:
            flask_restful.abort(404, message='Gifted sub {0} does not exist'.format(id))
        db.session.delete(gifted_sub)
        db.session.commit()
        return None, 204


class GiftedSubsResource(flask_restful.Resource):
    @flask_restful.marshal_with(gifted_sub_fields)
    def get(self):
        args = filter_parser.parse_args()
        q = db.session.query(GiftedSub)
        if args['filter']:
            q = q.filter(GiftedSub.giver.ilike('%{}%'.format(args['filter'])) |\
                         GiftedSub.receiver.ilike('%{}%'.format(args['filter'])))
        if args['older_than']:
            q = q.filter(GiftedSub.time < args['older_than'])
        if args['newer_than']:
            q = q.filter(GiftedSub.time > args['newer_than'])
        count = q.count()
        if args['sort_order'] == 'random':
            q = q.order_by(sqlalchemy.func.random())
        elif args['sort_by']:
            col = getattr(GiftedSub, args['sort_by'], None)
            if col:
                if args['sort_order']:
                    order_by = getattr(col, args['sort_order'], None)
                    if order_by:
                        q = q.order_by(order_by())
                else:
                    q = q.order_by(col)
        if args['page_size']:
            q = q.limit(args['page_size'])
        if args['page_number'] and args['page_size']:
            q = q.offset(args['page_number'] * args['page_size'])
        gifted_subs = q.all()
        return gifted_subs, 200, {'X-Total-Count': count}

    @flask_login.login_required
    @flask_restful.marshal_with(gifted_sub_fields)
    def post(self):
        args = gifted_sub_parser.parse_args()
        if not args['giver']:
            flask_restful.abort(400, message='Missing required parameter giver')
        if not args['receiver']:
            flask_restful.abort(400, message='Missing required parameter receiver')
        if not args['time']:
            flask_restful.abort(400, message='Missing required parameter time')
        now = sqlalchemy.func.now()
        q = db.session.query(GiftedSub).filter((GiftedSub.giver == args['giver']) &\
                (GiftedSub.receiver == args['receiver']) &\
                (sqlalchemy.func.DATE(GiftedSub.time) == args['time'].date()))
        gifted_sub = q.first()
        if not gifted_sub:
            gifted_sub = GiftedSub(created_at=now)
        gifted_sub.giver = args['giver']
        gifted_sub.receiver = args['receiver']
        gifted_sub.time = args['time']
        gifted_sub.updated_at = now
        db.session.add(gifted_sub)
        db.session.commit()
        url = api.url_for(GiftedSubResource, id=gifted_sub.id, _external=True, _scheme='https')
        return gifted_sub, 201, {'Location': url}


api.add_resource(RegularSubResource, '/regular-subs/<int:id>')
api.add_resource(RegularSubsResource, '/regular-subs')
api.add_resource(GiftedSubResource, '/gifted-subs/<int:id>')
api.add_resource(GiftedSubsResource, '/gifted-subs')


if __name__ == '__main__':
    app.run(host='0.0.0.0', threaded=True, debug=False)