Permalink
Fetching contributors…
Cannot retrieve contributors at this time
67 lines (41 sloc) 1.86 KB
from sqlalchemy import BigInteger, Column, String
from tornado.gen import coroutine
from tornado.ioloop import IOLoop
from tornado.options import define, options, parse_command_line
from tornado.web import Application, RequestHandler
from tornado_sqlalchemy import (SessionMixin, as_future, declarative_base,
make_session_factory)
DeclarativeBase = declarative_base()
define('database-url', type=str, help='Database URL')
class User(DeclarativeBase):
__tablename__ = 'users'
id = Column(BigInteger, primary_key=True)
username = Column(String(255), unique=True)
class SynchronousRequestHandler(SessionMixin, RequestHandler):
def get(self):
with self.make_session() as session:
count = session.query(User).count()
# OR count = self.session.query(User).count()
self.write('{} users so far!'.format(count))
class GenCoroutinesRequestHandler(SessionMixin, RequestHandler):
@coroutine
def get(self):
with self.make_session() as session:
count = yield as_future(session.query(User).count)
self.write('{} users so far!'.format(count))
class NativeCoroutinesRequestHandler(SessionMixin, RequestHandler):
async def get(self):
with self.make_session() as session:
count = await as_future(session.query(User).count)
self.write('{} users so far!'.format(count))
if __name__ == '__main__':
parse_command_line()
assert options.database_url, "Need a database URL"
session_factory = make_session_factory(options.database_url)
Application([
(r'/sync', SynchronousRequestHandler),
(r'/gen-coroutines', GenCoroutinesRequestHandler),
(r'/native-coroutines', NativeCoroutinesRequestHandler),
], session_factory=session_factory).listen(8888)
print('Listening on port 8888')
IOLoop.current().start()