ladylilia.com/quotes-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, Quote


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)


quote_fields = {
    'id': flask_restful.fields.Integer(),
    'date': flask_restful.fields.DateTime(dt_format='iso8601'),
    'game': flask_restful.fields.String(),
    'text': flask_restful.fields.String(),
    'created_at': flask_restful.fields.DateTime(dt_format='iso8601'),
    'updated_at': flask_restful.fields.DateTime(dt_format='iso8601'),
}


quote_parser = flask_restful.reqparse.RequestParser()
quote_parser.add_argument('id', type=int)
quote_parser.add_argument('date', type=flask_restful.inputs.date, required=True)
quote_parser.add_argument('game', type=str, required=True)
quote_parser.add_argument('text', type=str, required=True)


filter_parser = flask_restful.reqparse.RequestParser()
filter_parser.add_argument('filter', type=str)
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-Quotes-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 QuoteResource(flask_restful.Resource):
    @flask_restful.marshal_with(quote_fields)
    def get(self, id):
        q = db.session.query(Quote).filter(Quote.id == id)
        quote = q.first()
        if not quote:
            flask_restful.abort(404, message='Quote {0} does not exist'.format(id))
        return quote, 200

    @flask_login.login_required
    @flask_restful.marshal_with(quote_fields)
    def put(self, id):
        args = quote_parser.parse_args()
        now = sqlalchemy.func.now()
        q = db.session.query(Quote).filter(Quote.id == id)
        quote = q.first()
        if not quote:
            quote = Quote(id=id, created_at=now)
        quote.date = args['date']
        quote.game = args['game']
        quote.text = args['text']
        quote.updated_at = now
        db.session.add(quote)
        db.session.commit()
        return quote, 200

    @flask_login.login_required
    def delete(self, id):
        q = db.session.query(Quote).filter(Quote.id == id)
        quote = q.first()
        if not quote:
            flask_restful.abort(404, message='Quote {0} does not exist'.format(id))
        db.session.delete(quote)
        db.session.commit()
        q = db.session.query(Quote).filter(Quote.id > id)
        for quote in q.all():
            quote.id -= 1
            db.session.commit()
        return None, 204


class QuotesResource(flask_restful.Resource):
    @flask_restful.marshal_with(quote_fields)
    def get(self):
        args = filter_parser.parse_args()
        q = db.session.query(Quote)
        if args['filter']:
            q = q.filter(Quote.text.ilike('%{}%'.format(args['filter'])))
        count = q.count()
        if args['sort_order'] == 'random':
            q = q.order_by(sqlalchemy.func.random())
        elif args['sort_by']:
            col = getattr(Quote, 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'])
        quotes = q.all()
        return quotes, 200, {'X-Total-Count': count}

    @flask_login.login_required
    @flask_restful.marshal_with(quote_fields)
    def post(self):
        args = quote_parser.parse_args()
        if not args['id']:
            flask_restful.abort(400, message='Missing required parameter id')
        now = sqlalchemy.func.now()
        q = db.session.query(Quote).filter(Quote.id == args['id'])
        quote = q.first()
        if not quote:
            quote = Quote(id=args['id'], created_at=now)
        quote.date = args['date']
        quote.game = args['game']
        quote.text = args['text']
        quote.updated_at = now
        db.session.add(quote)
        db.session.commit()
        url = api.url_for(QuoteResource, id=quote.id, _external=True, _scheme='https')
        return quote, 201, {'Location': url}


api.add_resource(QuoteResource, '/quotes/<int:id>')
api.add_resource(QuotesResource, '/quotes')


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