main.py 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. import os
  2. import multiprocessing
  3. import asyncpg
  4. from aiohttp import web
  5. from sqlalchemy.engine.url import URL
  6. from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
  7. from .views import (
  8. json,
  9. single_database_query_orm,
  10. multiple_database_queries_orm,
  11. fortunes,
  12. updates,
  13. plaintext,
  14. single_database_query_raw,
  15. multiple_database_queries_raw,
  16. fortunes_raw,
  17. updates_raw,
  18. )
  19. CONNECTION_ORM = os.getenv('CONNECTION', 'ORM').upper() == 'ORM'
  20. def pg_dsn(dialect=None) -> str:
  21. """
  22. :return: DSN url suitable for sqlalchemy and aiopg.
  23. """
  24. url = URL.create(
  25. database='hello_world',
  26. password=os.getenv('PGPASS', 'benchmarkdbpass'),
  27. host='tfb-database',
  28. port='5432',
  29. username=os.getenv('PGUSER', 'benchmarkdbuser'),
  30. drivername='postgresql+{}'.format(dialect) if dialect else 'postgresql',
  31. )
  32. return url.render_as_string(hide_password=False)
  33. async def db_ctx(app: web.Application):
  34. # number of gunicorn workers = multiprocessing.cpu_count() as per gunicorn_conf.py
  35. # max_connections = 2000 as per toolset/setup/linux/databases/postgresql/postgresql.conf:64
  36. # give 10% leeway
  37. max_size = min(1800 / multiprocessing.cpu_count(), 160)
  38. max_size = max(int(max_size), 1)
  39. min_size = max(int(max_size / 2), 1)
  40. print(f'connection pool: min size: {min_size}, max size: {max_size}, orm: {CONNECTION_ORM}')
  41. if CONNECTION_ORM:
  42. dsn = pg_dsn('asyncpg')
  43. engine = create_async_engine(dsn, pool_size=max_size)
  44. app['db_session'] = async_sessionmaker(engine)
  45. else:
  46. dsn = pg_dsn()
  47. app['pg'] = await asyncpg.create_pool(dsn=dsn, min_size=min_size, max_size=max_size, loop=app.loop)
  48. yield
  49. if CONNECTION_ORM:
  50. await app['db_session'].dispose()
  51. else:
  52. await app['pg'].close()
  53. def setup_routes(app):
  54. if CONNECTION_ORM:
  55. app.router.add_get('/json', json)
  56. app.router.add_get('/db', single_database_query_orm)
  57. app.router.add_get('/queries/{queries:.*}', multiple_database_queries_orm)
  58. app.router.add_get('/fortunes', fortunes)
  59. app.router.add_get('/updates/{queries:.*}', updates)
  60. app.router.add_get('/plaintext', plaintext)
  61. else:
  62. app.router.add_get('/db', single_database_query_raw)
  63. app.router.add_get('/queries/{queries:.*}', multiple_database_queries_raw)
  64. app.router.add_get('/fortunes', fortunes_raw)
  65. app.router.add_get('/updates/{queries:.*}', updates_raw)
  66. def create_app():
  67. app = web.Application()
  68. app.cleanup_ctx.append(db_ctx)
  69. setup_routes(app)
  70. return app