SHARE
TWEET

Untitled

a guest Dec 13th, 2016 56 Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
  1. import logging
  2. import asyncio
  3. import aiomysql
  4. from aiohttp import web
  5. from aiomysql.cursors import DictCursor
  6.  
  7.  
  8. logging.basicConfig(level=logging.DEBUG)
  9.  
  10. async def index(request):
  11.     async with request.app["mysql"].acquire() as conn:
  12.         async with conn.cursor() as cur:
  13.             await cur.execute("SELECT * FROM my_table")
  14.             lines = await cur.fetchall()
  15.  
  16.     return web.Response(text='Hello Aiohttp!')
  17.  
  18. async def get_mysql_pool(loop):
  19.     pool = await aiomysql.create_pool(
  20.         host="localhost",
  21.         user="test",
  22.         password="test",
  23.         db="test",
  24.         cursorclass=DictCursor,
  25.         loop=loop
  26.     )
  27.  
  28.     return pool
  29.  
  30. if __name__ == "__main__":
  31.     loop = asyncio.get_event_loop()
  32.     mysql = loop.run_until_complete(get_mysql_pool(loop))
  33.     app = web.Application(loop=loop, debug=True)
  34.     app["mysql"] = mysql
  35.     app.router.add_get("/", index)
  36.     web.run_app(app)
RAW Paste Data
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy. OK, I Understand
 
Top