app_asgi.py 5.1 KB

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