A tornado server that creates an API and schema output automatically based on a SQLAlchemy model

tornado_sqlalchemy_auto_api.py

#!/usr/bin/env python3
# vim: set fileencoding=utf8 :

import tornado.ioloop
from tornado.options import options, define

from tornado_sqlalchemy import SQLAlchemy
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

import tornado_jsonapi.handlers
import tornado_jsonapi.resource

from alchemyjsonschema import SchemaFactory
from alchemyjsonschema import StructuralWalker
from tornado_sqlalchemy import SessionMixin

db = SQLAlchemy(url="sqlite:///db.db")


class SomeRequestHandler(SessionMixin, tornado.web.RequestHandler):
    def initialize(self, name: str, cls: bool = True) -> None:
        self._name = name
        self._cls = cls

    def get(self):
        # for actual table names:
        # tables = db.Model.metadata.tables.keys()
        factory = SchemaFactory(StructuralWalker)

        json = factory(self._cls)
        self.write(json)

        # # if we wanted to do this a little bit more dynamically...
        # engine = self.session.get_bind()
        # tables = engine.table_names()

        # # https://stackoverflow.com/a/26518401/193772
        # classes, models, table_names = [], [], []
        # for clazz in self.session.db.Model._decl_class_registry.values():
        #     try:
        #         table_names.append(clazz.__tablename__)
        #         classes.append(clazz)
        #     except:
        #         pass
        # for table in self.session.db.metadata.tables.items():
        #     if table[0] in table_names:
        #         models.append(classes[table_names.index(table[0])])

        # tables = self.session.db.Model._decl_class_registry
        # for c in tables.keys():
        #     v = db.Model._decl_class_registry[c]
        #     try:
        #         if c != slug and v.__tablename__ != slug:
        #             continue
        #     except:
        #         continue
        #         factory = SchemaFactory(StructuralWalker)

        #         json = factory(v)
        #         self.write(json)

        #     return

        # self.write('{}')
        # return


class Post(db.Model):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)
    author = Column(String)
    text = Column(String)


def main():
    define("debug", default=False, help="Run in debug mode")
    options.parse_command_line()
    settings = {}
    settings.update(options.group_dict(None))
    settings.update(tornado_jsonapi.handlers.not_found_handling_settings())

    apis = []
    for tableName in db.Model._decl_class_registry.keys():
        tableClass = db.Model._decl_class_registry[tableName]
        if hasattr(tableClass, "__tablename__"):
            apis.append(
                (
                    r"/api/schema/%s" % tableName.lower(),
                    SomeRequestHandler,
                    dict(name=tableName, cls=tableClass),
                )
            )
            apis.append(
                (
                    r"/api/%s/([^/]*)" % tableName.lower(),
                    tornado_jsonapi.handlers.APIHandler,
                    dict(
                        resource=tornado_jsonapi.resource.SQLAlchemyResource(
                            tableClass, sessionmaker(bind=db.engine)
                        )
                    ),
                )
            )

    # if we wanted the schemas to a bit more dynamic
    # handlers = [
    #     (
    #         r'/api/schemas/([\w\.]*)\/?',
    #         SomeRequestHandler
    #     )
    # ]
    # handlers.extend(apis)

    application = tornado.web.Application(apis, db=db, **settings)
    application.listen(8888)
    tornado.ioloop.IOLoop.current().start()


if __name__ == "__main__":
    main()