Skip to content

Instantly share code, notes, and snippets.

@twodayslate
Created January 10, 2020 22:33
Show Gist options
  • Save twodayslate/c780af714fc091e040acbf32cd82b6f2 to your computer and use it in GitHub Desktop.
Save twodayslate/c780af714fc091e040acbf32cd82b6f2 to your computer and use it in GitHub Desktop.
A tornado server that creates an API and schema output automatically based on a SQLAlchemy model
#!/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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment