main.py 2.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182
  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 AsyncSession, create_async_engine
  7. from sqlalchemy.orm import sessionmaker
  8. from .views import (
  9. json,
  10. single_database_query_orm,
  11. multiple_database_queries_orm,
  12. fortunes,
  13. updates,
  14. plaintext,
  15. single_database_query_raw,
  16. multiple_database_queries_raw,
  17. fortunes_raw,
  18. updates_raw,
  19. )
  20. CONNECTION_ORM = os.getenv('CONNECTION', 'ORM').upper() == 'ORM'
  21. def pg_dsn(dialect=None) -> str:
  22. """
  23. :return: DSN url suitable for sqlalchemy and aiopg.
  24. """
  25. return str(URL.create(
  26. database='hello_world',
  27. password=os.getenv('PGPASS', 'benchmarkdbpass'),
  28. host='tfb-database',
  29. port='5432',
  30. username=os.getenv('PGUSER', 'benchmarkdbuser'),
  31. drivername='postgresql+{}'.format(dialect) if dialect else 'postgresql',
  32. ))
  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, future=True, pool_size=max_size)
  44. app['db_session'] = sessionmaker(engine, class_=AsyncSession)
  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 not CONNECTION_ORM:
  50. await app['pg'].close()
  51. def setup_routes(app):
  52. if CONNECTION_ORM:
  53. app.router.add_get('/json', json)
  54. app.router.add_get('/db', single_database_query_orm)
  55. app.router.add_get('/queries/{queries:.*}', multiple_database_queries_orm)
  56. app.router.add_get('/fortunes', fortunes)
  57. app.router.add_get('/updates/{queries:.*}', updates)
  58. app.router.add_get('/plaintext', plaintext)
  59. else:
  60. app.router.add_get('/db', single_database_query_raw)
  61. app.router.add_get('/queries/{queries:.*}', multiple_database_queries_raw)
  62. app.router.add_get('/fortunes', fortunes_raw)
  63. app.router.add_get('/updates/{queries:.*}', updates_raw)
  64. def create_app():
  65. app = web.Application()
  66. app.cleanup_ctx.append(db_ctx)
  67. setup_routes(app)
  68. return app