app_asgi.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208
  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. worlds = []
  87. async with pool.acquire() as connection:
  88. rows = await connection.fetchmany(SQL_SELECT, [(v,) for v in row_ids])
  89. worlds = [{'id': row_id, 'randomNumber': number[0]} for row_id, number in zip(row_ids, rows)]
  90. await send(JSON_RESPONSE)
  91. await send({
  92. 'type': 'http.response.body',
  93. 'body': json_dumps(worlds),
  94. 'more_body': False
  95. })
  96. async def route_fortunes(scope, receive, send):
  97. async with pool.acquire() as connection:
  98. fortunes = await connection.fetch('SELECT * FROM Fortune')
  99. fortunes.append(ROW_ADD)
  100. fortunes.sort(key=key)
  101. content = template.render(fortunes=fortunes).encode('utf-8')
  102. await send(HTML_RESPONSE)
  103. await send({
  104. 'type': 'http.response.body',
  105. 'body': content,
  106. 'more_body': False
  107. })
  108. async def route_updates(scope, receive, send):
  109. num_queries = get_num_queries(scope)
  110. updates = list(zip(
  111. sample(range(1, 10000), num_queries),
  112. sorted(sample(range(1, 10000), num_queries))
  113. ))
  114. worlds = [{'id': row_id, 'randomNumber': number} for row_id, number in updates]
  115. async with pool.acquire() as connection:
  116. await connection.executemany(SQL_SELECT, [(i[0],) for i in updates])
  117. await connection.executemany(SQL_UPDATE, updates)
  118. await send(JSON_RESPONSE)
  119. await send({
  120. 'type': 'http.response.body',
  121. 'body': json_dumps(worlds),
  122. 'more_body': False
  123. })
  124. async def route_plaintext(scope, receive, send):
  125. await send(PLAINTEXT_RESPONSE)
  126. await send({
  127. 'type': 'http.response.body',
  128. 'body': b'Hello, world!',
  129. 'more_body': False
  130. })
  131. async def handle_404(scope, receive, send):
  132. await send(PLAINTEXT_RESPONSE)
  133. await send({
  134. 'type': 'http.response.body',
  135. 'body': b'Not found',
  136. 'more_body': False
  137. })
  138. routes = {
  139. '/json': route_json,
  140. '/db': route_db,
  141. '/queries': route_queries,
  142. '/fortunes': route_fortunes,
  143. '/updates': route_updates,
  144. '/plaintext': route_plaintext
  145. }
  146. class App:
  147. __slots__ = ["_handler"]
  148. def __init__(self):
  149. self._handler = self._lifespan
  150. def __call__(self, scope, receive, send):
  151. return self._handler(scope, receive, send)
  152. async def _lifespan(self, scope, receive, send):
  153. if scope['type'] == 'lifespan':
  154. message = await receive()
  155. if message['type'] == 'lifespan.startup':
  156. await pg_setup()
  157. self._handler = self._asgi
  158. await send({'type': 'lifespan.startup.complete'})
  159. def _asgi(self, scope, receive, send):
  160. handler = routes.get(scope['path'], handle_404)
  161. return handler(scope, receive, send)
  162. main = App()