TechEmpower.FrameworkBenchmarks/frameworks/Python/fastwsgi/app-asgi.py

import os
import sys
import asyncio
import asyncpg
import jinja2
import random
from operator import itemgetter
from urllib.parse import parse_qs
import asyncache
import cachetools

try:
    from ujson import dumps as jsonify
except:
    from json import dumps as jsonify


db_pool = None

async def db_setup():
    global db_pool
    db_pool = await asyncpg.create_pool(
        user=os.getenv('PGUSER', 'benchmarkdbuser'),
        password=os.getenv('PGPASS', 'benchmarkdbpass'),
        database='hello_world',
        host='tfb-database',
        port=5432
    )

READ_ROW_SQL = 'SELECT "randomnumber", "id" FROM "world" WHERE id = $1'
WRITE_ROW_SQL = 'UPDATE "world" SET "randomnumber"=$1 WHERE id=$2'
ADDITIONAL_ROW = [0, 'Additional fortune added at request time.']

JSON_RESPONSE = {
    'type': 'http.response.start',
    'status': 200,
    'headers': [
        [b'Content-Type', b'application/json'],
    ]
}

HTML_RESPONSE = {
    'type': 'http.response.start',
    'status': 200,
    'headers': [
        [b'Content-Type', b'text/html; charset=utf-8'],
    ]
}

PLAINTEXT_RESPONSE = {
    'type': 'http.response.start',
    'status': 200,
    'headers': [
        [b'Content-Type', b'text/plain; charset=utf-8'],
    ]
}


def get_num_queries(scope, name = b'queries'):
    try:
        query_string = scope['query_string']
        query_count = int(parse_qs(query_string)[name][0])
    except (KeyError, IndexError, ValueError):
        return 1
    if query_count < 1:
        return 1
    if query_count > 500:
        return 500
    return query_count


async def json_serialization(scope, receive, send):
    content = jsonify( {'message': 'Hello, world!'} )
    await send(JSON_RESPONSE)
    await send({
        'type': 'http.response.body',
        'body': content.encode('utf8'),
        'more_body': False
    })


async def single_database_query(scope, receive, send):
    row_id = random.randint(1, 10000)
    db_conn = await db_pool.acquire()
    try:
        number = await db_conn.fetchval(READ_ROW_SQL, row_id)
        world = {'id': row_id, 'randomNumber': number}
    finally:
        await db_pool.release(db_conn)

    content = jsonify(world)
    await send(JSON_RESPONSE)
    await send({
        'type': 'http.response.body',
        'body': content.encode('utf8'),
        'more_body': False
    })


async def multiple_database_queries(scope, receive, send):
    num_queries = get_num_queries(scope)
    row_ids = random.sample(range(1, 10000), num_queries)
    worlds = [ ]

    db_conn = await db_pool.acquire()
    try:
        statement = await db_conn.prepare(READ_ROW_SQL)
        for row_id in row_ids:
            number = await statement.fetchval(row_id)
            worlds.append( {'id': row_id, 'randomNumber': number} )
    finally:
        await db_pool.release(db_conn)

    content = jsonify(worlds)
    await send(JSON_RESPONSE)
    await send({
        'type': 'http.response.body',
        'body': content.encode('utf8'),
        'more_body': False
    })


_get_item1 = itemgetter(1)
fortunes_template = None
fn = os.path.join('templates', 'fortune.html')
with open(fn, 'r') as file:
    text = file.read()
    fortunes_template = jinja2.Template(text)


async def fortunes(scope, receive, send):
    db_conn = await db_pool.acquire()
    try:
        fortunes = await db_conn.fetch('SELECT * FROM Fortune')
    finally:
        await db_pool.release(db_conn)

    fortunes.append(ADDITIONAL_ROW)
    fortunes.sort(key = _get_item1)
    content = fortunes_template.render(fortunes=fortunes)
    await send(HTML_RESPONSE)
    await send({
        'type': 'http.response.body',
        'body': content.encode('utf8'),
        'more_body': False
    })


async def database_updates(scope, receive, send):
    num_queries = get_num_queries(scope)
    ids = sorted(random.sample(range(1, 10000 + 1), num_queries))
    numbers = sorted(random.sample(range(1, 10000), num_queries))
    updates = list(zip(ids, numbers))
    
    worlds = [ {"id": row_id, "randomNumber": number} for row_id, number in updates ]

    db_conn = await db_pool.acquire()
    try:
        statement = await db_conn.prepare(READ_ROW_SQL)
        for row_id, _ in updates:
            await statement.fetchval(row_id)
        await db_conn.executemany(WRITE_ROW_SQL, updates)
    finally:
        await db_pool.release(db_conn)

    content = jsonify(worlds)
    await send(JSON_RESPONSE)
    await send({
        'type': 'http.response.body',
        'body': content.encode('utf8'),
        'more_body': False
    })


from asyncache import cached
from cachetools.keys import hashkey

@cached(cache={}, key=lambda stmt, id: hashkey(id))
async def get_cached_world(stmt, id):
    result = await stmt.fetchrow(id)
    return {'id': result[1], 'randomNumber': result[0]}

async def cached_queries(scope, receive, send):
    count = get_num_queries(scope, b'count')
    row_ids = random.sample(range(1, 10000 + 1), count)
    
    db_conn = await db_pool.acquire()
    try:
        statement = await db_conn.prepare(READ_ROW_SQL)
        worlds = [ await get_cached_world(statement, id) for id in row_ids ]
    finally:
        await db_pool.release(db_conn)

    content = jsonify(worlds)
    await send(JSON_RESPONSE)
    await send({
        'type': 'http.response.body',
        'body': content.encode('utf8'),
        'more_body': False
    })


async def plaintext(scope, receive, send):
    await send(PLAINTEXT_RESPONSE)
    await send({
        'type': 'http.response.body',
        'body': b'Hello, world!',
        'more_body': False
    })


async def handle_404(scope, receive, send):
    await send(PLAINTEXT_RESPONSE)
    await send({
        'type': 'http.response.body',
        'body': b'Not found',
        'more_body': False
    })


routes = {
    '/json': json_serialization,
    '/db': single_database_query,
    '/queries': multiple_database_queries,
    '/fortunes': fortunes,
    '/updates': database_updates,
    '/cached-queries': cached_queries,
    '/plaintext': plaintext,
}


async def app(scope, receive, send):
    if scope['type'] == 'lifespan':
        global db_pool
        while True:
            message = await receive()
            if message['type'] == 'lifespan.startup':
                await db_setup()
                await send({'type': 'lifespan.startup.complete'})
            elif message['type'] == 'lifespan.shutdown':
                db_pool.close()
                await send({'type': 'lifespan.shutdown.complete'})
                return            
    else:
        path = scope['path']
        app_handler = routes.get(path, handle_404)
        await app_handler(scope, receive, send)

# -----------------------------------------------------------------------------------------------------

if __name__ == "__main__":
    import multiprocessing
    import fastwsgi

    _is_travis = os.environ.get('TRAVIS') == 'true'

    workers = int(multiprocessing.cpu_count())
    if _is_travis:
        workers = 2

    host = '0.0.0.0'
    port = 3000

    def run_app():
        loop = asyncio.get_event_loop()
        loop.run_until_complete(db_setup())
        fastwsgi.server.backlog = 4096
        fastwsgi.run(app, host, port, loglevel=2)

    def create_fork():
        n = os.fork()
        # n greater than 0 means parent process
        if not n > 0:
            run_app()

    # fork limiting the cpu count - 1
    for i in range(1, workers):
        create_fork()

    run_app()  # run app on the main process too :)