ladylilia.com/instagram-api/app.py

import logging
import os

import flask
import flask_apscheduler
import flask_restful
import flask_restful.fields
import flask_restful.reqparse
import sqlalchemy
import sqlalchemy.engine

from db import db, Medium


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'),
    SCHEDULER_TIMEZONE='UTC',
    SCHEDULER_JOBS=[
        dict(id='sync_media',
             func='sync:Sync.sync_media',
             args=(app, db),
             max_instances=1,
             trigger='interval',
             seconds=300)])

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)

scheduler = flask_apscheduler.APScheduler()
scheduler.init_app(app)

api = flask_restful.Api(app)


medium_fields = {
    'id': flask_restful.fields.Integer(),
    'typename': flask_restful.fields.String(),
    'caption': flask_restful.fields.String(),
    'shortcode': flask_restful.fields.String(),
    'taken_at': flask_restful.fields.DateTime(dt_format='iso8601'),
    'width': flask_restful.fields.Integer(),
    'height': flask_restful.fields.Integer(),
    'display_url': flask_restful.fields.String(),
    'thumbnail_url': flask_restful.fields.String(),
    'likes': flask_restful.fields.Integer(),
    'owner_id': flask_restful.fields.Integer(),
    'owner_username': flask_restful.fields.String(),
    'owner_profile_pic_url': flask_restful.fields.String(),
}


filter_parser = flask_restful.reqparse.RequestParser()
filter_parser.add_argument('filter', type=str)
filter_parser.add_argument('type', 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)


class MediumResource(flask_restful.Resource):
    @flask_restful.marshal_with(medium_fields)
    def get(self, id):
        q = db.session.query(Medium).filter(Medium.id == id)
        medium = q.first()
        if not medium:
            flask_restful.abort(404, message='Medium {0} does not exist'.format(id))
        return medium, 200


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


api.add_resource(MediumResource, '/media/<int:id>')
api.add_resource(MediaResource, '/media')


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