app_asgi.py 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183
  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. asyncio.get_event_loop().run_until_complete(pg_setup())
  49. def get_num_queries(scope):
  50. try:
  51. query_string = scope['query_string']
  52. query_count = int(parse_qs(query_string)[b'queries'][0])
  53. except (KeyError, IndexError, ValueError):
  54. return 1
  55. if query_count < 1:
  56. return 1
  57. if query_count > 500:
  58. return 500
  59. return query_count
  60. async def route_json(scope, receive, send):
  61. await send(JSON_RESPONSE)
  62. await send({
  63. 'type': 'http.response.body',
  64. 'body': json_dumps({'message': 'Hello, world!'}),
  65. 'more_body': False
  66. })
  67. async def route_db(scope, receive, send):
  68. row_id = randint(1, 10000)
  69. async with pool.acquire() as connection:
  70. number = await connection.fetchval(SQL_SELECT, row_id)
  71. await send(JSON_RESPONSE)
  72. await send({
  73. 'type': 'http.response.body',
  74. 'body': json_dumps({'id': row_id, 'randomNumber': number}),
  75. 'more_body': False
  76. })
  77. async def route_queries(scope, receive, send):
  78. num_queries = get_num_queries(scope)
  79. row_ids = sample(range(1, 10000), num_queries)
  80. worlds = []
  81. async with pool.acquire() as connection:
  82. statement = await connection.prepare(SQL_SELECT)
  83. for row_id in row_ids:
  84. number = await statement.fetchval(row_id)
  85. worlds.append({'id': row_id, 'randomNumber': number})
  86. await send(JSON_RESPONSE)
  87. await send({
  88. 'type': 'http.response.body',
  89. 'body': json_dumps(worlds),
  90. 'more_body': False
  91. })
  92. async def route_fortunes(scope, receive, send):
  93. async with pool.acquire() as connection:
  94. fortunes = await connection.fetch('SELECT * FROM Fortune')
  95. fortunes.append(ROW_ADD)
  96. fortunes.sort(key=key)
  97. content = template.render(fortunes=fortunes).encode('utf-8')
  98. await send(HTML_RESPONSE)
  99. await send({
  100. 'type': 'http.response.body',
  101. 'body': content,
  102. 'more_body': False
  103. })
  104. async def route_updates(scope, receive, send):
  105. num_queries = get_num_queries(scope)
  106. updates = list(zip(
  107. sample(range(1, 10000), num_queries),
  108. sorted(sample(range(1, 10000), num_queries))
  109. ))
  110. worlds = [{'id': row_id, 'randomNumber': number} for row_id, number in updates]
  111. async with pool.acquire() as connection:
  112. statement = await connection.prepare(SQL_SELECT)
  113. for row_id, _ in updates:
  114. await statement.fetchval(row_id)
  115. await connection.executemany(SQL_UPDATE, updates)
  116. await send(JSON_RESPONSE)
  117. await send({
  118. 'type': 'http.response.body',
  119. 'body': json_dumps(worlds),
  120. 'more_body': False
  121. })
  122. async def route_plaintext(scope, receive, send):
  123. await send(PLAINTEXT_RESPONSE)
  124. await send({
  125. 'type': 'http.response.body',
  126. 'body': b'Hello, world!',
  127. 'more_body': False
  128. })
  129. async def handle_404(scope, receive, send):
  130. await send(PLAINTEXT_RESPONSE)
  131. await send({
  132. 'type': 'http.response.body',
  133. 'body': b'Not found',
  134. 'more_body': False
  135. })
  136. routes = {
  137. '/json': route_json,
  138. '/db': route_db,
  139. '/queries': route_queries,
  140. '/fortunes': route_fortunes,
  141. '/updates': route_updates,
  142. '/plaintext': route_plaintext
  143. }
  144. def main(scope, receive, send):
  145. handler = routes.get(scope['path'], handle_404)
  146. return handler(scope, receive, send)