__init__.py 5.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102
  1. import logging
  2. import asyncio
  3. import os
  4. import aiopg
  5. import jinja2
  6. import psycopg2.extras
  7. import asyncio_redis
  8. from asyncio_redis.protocol import HiRedisProtocol
  9. import aiohttp.web
  10. import aiohttp_jinja2
  11. import aiomysql
  12. import api_hour
  13. from . import endpoints
  14. LOG = logging.getLogger(__name__)
  15. class Container(api_hour.Container):
  16. def __init__(self, *args, **kwargs):
  17. super().__init__(*args, **kwargs)
  18. # Servers
  19. self.servers['http'] = aiohttp.web.Application(loop=kwargs['loop'])
  20. aiohttp_jinja2.setup(self.servers['http'], loader=jinja2.PackageLoader('hello'))
  21. self.servers['http'].ah_container = self # keep a reference to Container
  22. # routes
  23. self.servers['http'].router.add_route('GET', '/json', endpoints.world.json)
  24. self.servers['http'].router.add_route('GET', '/db', endpoints.world.db)
  25. self.servers['http'].router.add_route('GET', '/db_redis', endpoints.world.db_redis)
  26. self.servers['http'].router.add_route('GET', '/db_mysql', endpoints.world.db_mysql)
  27. self.servers['http'].router.add_route('GET', '/queries', endpoints.world.queries)
  28. self.servers['http'].router.add_route('GET', '/queries_redis', endpoints.world.queries_redis)
  29. self.servers['http'].router.add_route('GET', '/queries_mysql', endpoints.world.queries_mysql)
  30. self.servers['http'].router.add_route('GET', '/fortunes', endpoints.world.fortunes)
  31. self.servers['http'].router.add_route('GET', '/fortunes_redis', endpoints.world.fortunes_redis)
  32. self.servers['http'].router.add_route('GET', '/fortunes_mysql', endpoints.world.fortunes_mysql)
  33. self.servers['http'].router.add_route('GET', '/updates', endpoints.world.updates)
  34. self.servers['http'].router.add_route('GET', '/updates_redis', endpoints.world.updates_redis)
  35. self.servers['http'].router.add_route('GET', '/updates_mysql', endpoints.world.updates_mysql)
  36. self.servers['http'].router.add_route('GET', '/plaintext', endpoints.world.plaintext)
  37. def make_servers(self):
  38. return [self.servers['http'].make_handler(logger=self.worker.log,
  39. debug=False,
  40. keep_alive=0,
  41. access_log=None,
  42. access_log_format=self.worker.cfg.access_log_format)]
  43. @asyncio.coroutine
  44. def start(self):
  45. yield from super().start()
  46. LOG.info('Starting engines...')
  47. self.engines['pg'] = self.loop.create_task(aiopg.create_pool(host=os.environ.get('DBHOST', self.config['engines']['pg']['host']),
  48. port=int(self.config['engines']['pg']['port']),
  49. sslmode='disable',
  50. dbname=self.config['engines']['pg']['dbname'],
  51. user=self.config['engines']['pg']['user'],
  52. password=self.config['engines']['pg']['password'],
  53. cursor_factory=psycopg2.extras.RealDictCursor,
  54. minsize=int(self.config['engines']['pg']['minsize']),
  55. maxsize=int(self.config['engines']['pg']['maxsize']),
  56. loop=self.loop))
  57. self.engines['mysql'] = self.loop.create_task(aiomysql.create_pool(
  58. host=self.config['engines']['mysql']['host'],
  59. port=self.config['engines']['mysql']['port'],
  60. user=self.config['engines']['mysql']['user'],
  61. password=self.config['engines']['mysql']['pwd'],
  62. db=self.config['engines']['mysql']['db'],
  63. minsize=int(self.config['engines']['mysql']['minsize']),
  64. maxsize=int(self.config['engines']['mysql']['maxsize']),
  65. cursorclass=aiomysql.DictCursor,
  66. loop=self.loop))
  67. yield from asyncio.wait([self.engines['pg']], return_when=asyncio.ALL_COMPLETED)
  68. self.engines['redis'] = yield from asyncio_redis.Pool.create(host=self.config['engines']['redis']['host'],
  69. port=self.config['engines']['redis']['port'],
  70. poolsize=self.config['engines']['redis']['poolsize'],
  71. loop=self.loop,
  72. protocol_class=HiRedisProtocol)
  73. LOG.info('All engines ready !')
  74. @asyncio.coroutine
  75. def stop(self):
  76. LOG.info('Stopping engines...')
  77. if 'pg' in self.engines:
  78. if self.engines['pg'].done():
  79. self.engines['pg'].result().terminate()
  80. yield from self.engines['pg'].result().wait_closed()
  81. else:
  82. yield from self.engines['pg'].cancel()
  83. if 'mysql' in self.engines:
  84. if self.engines['mysql'].done():
  85. self.engines['mysql'].result().close()
  86. yield from self.engines['mysql'].result().wait_closed()
  87. else:
  88. yield from self.engines['mysql'].cancel()
  89. if 'redis' in self.engines:
  90. self.engines['redis'].close()
  91. yield from asyncio.sleep(1) # wait redis close connection
  92. LOG.info('All engines stopped !')
  93. yield from super().stop()