app_asgi.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207
  1. import os
  2. from operator import itemgetter
  3. from pathlib import Path
  4. from random import randint, sample
  5. from urllib.parse import parse_qs
  6. import asyncpg
  7. import jinja2
  8. import orjson
  9. PG_POOL_SIZE = 4
  10. class NoResetConnection(asyncpg.Connection):
  11. __slots__ = ()
  12. def get_reset_query(self):
  13. return ""
  14. async def pg_setup():
  15. global pool
  16. pool = await asyncpg.create_pool(
  17. user=os.getenv('PGUSER', 'benchmarkdbuser'),
  18. password=os.getenv('PGPASS', 'benchmarkdbpass'),
  19. database='hello_world',
  20. host='tfb-database',
  21. port=5432,
  22. min_size=PG_POOL_SIZE,
  23. max_size=PG_POOL_SIZE,
  24. connection_class=NoResetConnection,
  25. )
  26. SQL_SELECT = 'SELECT "randomnumber", "id" FROM "world" WHERE id = $1'
  27. SQL_UPDATE = 'UPDATE "world" SET "randomnumber"=$1 WHERE id=$2'
  28. ROW_ADD = [0, 'Additional fortune added at request time.']
  29. JSON_RESPONSE = {
  30. 'type': 'http.response.start',
  31. 'status': 200,
  32. 'headers': [
  33. [b'content-type', b'application/json'],
  34. ]
  35. }
  36. HTML_RESPONSE = {
  37. 'type': 'http.response.start',
  38. 'status': 200,
  39. 'headers': [
  40. [b'content-type', b'text/html; charset=utf-8'],
  41. ]
  42. }
  43. PLAINTEXT_RESPONSE = {
  44. 'type': 'http.response.start',
  45. 'status': 200,
  46. 'headers': [
  47. [b'content-type', b'text/plain; charset=utf-8'],
  48. ]
  49. }
  50. pool = None
  51. key = itemgetter(1)
  52. json_dumps = orjson.dumps
  53. with Path('templates/fortune.html').open('r') as f:
  54. template = jinja2.Template(f.read())
  55. def get_num_queries(scope):
  56. try:
  57. query_string = scope['query_string']
  58. query_count = int(parse_qs(query_string)[b'queries'][0])
  59. except (KeyError, IndexError, ValueError):
  60. return 1
  61. if query_count < 1:
  62. return 1
  63. if query_count > 500:
  64. return 500
  65. return query_count
  66. async def route_json(scope, receive, send):
  67. await send(JSON_RESPONSE)
  68. await send({
  69. 'type': 'http.response.body',
  70. 'body': json_dumps({'message': 'Hello, world!'}),
  71. 'more_body': False
  72. })
  73. async def route_db(scope, receive, send):
  74. row_id = randint(1, 10000)
  75. async with pool.acquire() as connection:
  76. number = await connection.fetchval(SQL_SELECT, row_id)
  77. await send(JSON_RESPONSE)
  78. await send({
  79. 'type': 'http.response.body',
  80. 'body': json_dumps({'id': row_id, 'randomNumber': number}),
  81. 'more_body': False
  82. })
  83. async def route_queries(scope, receive, send):
  84. num_queries = get_num_queries(scope)
  85. row_ids = sample(range(1, 10000), num_queries)
  86. async with pool.acquire() as connection:
  87. rows = await connection.fetchmany(SQL_SELECT, [(v,) for v in row_ids])
  88. worlds = [{'id': row_id, 'randomNumber': number[0]} for row_id, number in zip(row_ids, rows)]
  89. await send(JSON_RESPONSE)
  90. await send({
  91. 'type': 'http.response.body',
  92. 'body': json_dumps(worlds),
  93. 'more_body': False
  94. })
  95. async def route_fortunes(scope, receive, send):
  96. async with pool.acquire() as connection:
  97. fortunes = await connection.fetch('SELECT * FROM Fortune')
  98. fortunes.append(ROW_ADD)
  99. fortunes.sort(key=key)
  100. content = template.render(fortunes=fortunes).encode('utf-8')
  101. await send(HTML_RESPONSE)
  102. await send({
  103. 'type': 'http.response.body',
  104. 'body': content,
  105. 'more_body': False
  106. })
  107. async def route_updates(scope, receive, send):
  108. num_queries = get_num_queries(scope)
  109. updates = list(zip(
  110. sample(range(1, 10000), num_queries),
  111. sorted(sample(range(1, 10000), num_queries))
  112. ))
  113. worlds = [{'id': row_id, 'randomNumber': number} for row_id, number in updates]
  114. async with pool.acquire() as connection:
  115. await connection.executemany(SQL_SELECT, [(i[0],) for i in updates])
  116. await connection.executemany(SQL_UPDATE, updates)
  117. await send(JSON_RESPONSE)
  118. await send({
  119. 'type': 'http.response.body',
  120. 'body': json_dumps(worlds),
  121. 'more_body': False
  122. })
  123. async def route_plaintext(scope, receive, send):
  124. await send(PLAINTEXT_RESPONSE)
  125. await send({
  126. 'type': 'http.response.body',
  127. 'body': b'Hello, world!',
  128. 'more_body': False
  129. })
  130. async def handle_404(scope, receive, send):
  131. await send(PLAINTEXT_RESPONSE)
  132. await send({
  133. 'type': 'http.response.body',
  134. 'body': b'Not found',
  135. 'more_body': False
  136. })
  137. routes = {
  138. '/json': route_json,
  139. '/db': route_db,
  140. '/queries': route_queries,
  141. '/fortunes': route_fortunes,
  142. '/updates': route_updates,
  143. '/plaintext': route_plaintext
  144. }
  145. class App:
  146. __slots__ = ["_handler"]
  147. def __init__(self):
  148. self._handler = self._lifespan
  149. def __call__(self, scope, receive, send):
  150. return self._handler(scope, receive, send)
  151. async def _lifespan(self, scope, receive, send):
  152. if scope['type'] == 'lifespan':
  153. message = await receive()
  154. if message['type'] == 'lifespan.startup':
  155. await pg_setup()
  156. self._handler = self._asgi
  157. await send({'type': 'lifespan.startup.complete'})
  158. def _asgi(self, scope, receive, send):
  159. handler = routes.get(scope['path'], handle_404)
  160. return handler(scope, receive, send)
  161. main = App()