importasyncioimporttimefromaiohttpimportwebfromsqlalchemyimportColumn,Integer,Stringfromsqlalchemy.ext.declarativeimportdeclarative_baseBase=declarative_base()classUser(Base):__tablename__='tbl_user'id=Column(Integer,primary_key=True)name=Column(String(50),nullable=True)@classmethoddeflong_sync(cls):# Some query on dbtime.sleep(5)return'Data from db'@asyncio.coroutinedefsync_db(func):loop=asyncio.get_event_loop()result=loop.run_in_executor(None,func)returnresult@asyncio.coroutinedefhandle(request):data=yield fromsync_db(User.long_sync)text='Hello, '+datareturnweb.Response(body=text.encode('utf-8'))@asyncio.coroutinedefinit(loop):app=web.Application(loop=loop)app.router.add_route('GET','/',handle)srv=yield fromloop.create_server(app.make_handler(),'127.0.0.1',8080)print("Server started at http://127.0.0.1:8080")returnsrvif__name__=='__main__':loop=asyncio.get_event_loop()loop.run_until_complete(init(loop))try:loop.run_forever()exceptKeyboardInterrupt:pass
Comments (0)
HTTPSSSH
You can clone a snippet to your computer for local editing.
Learn more.