Running blocking tasks in a thread with aiohttp

aiohttp_threading.py

import asyncio
from aiohttp import web
from concurrent.futures import ThreadPoolExecutor
import time


def blocking_func(seconds: int) -> int:
    time.sleep(seconds)
    return seconds


async def view_page(request: web.Request):
    seconds = int(request.query.get("seconds", 5))
    executor = request.app["executor"]
    loop = asyncio.get_event_loop()
    task = loop.run_in_executor(executor, blocking_func, seconds)
    completed, pending = await asyncio.wait([task])
    result = task.result()
    return web.Response(text=f"Waited {result} second(s).")


def create_app():
    app = web.Application()
    app.add_routes([web.get("/", view_page)])
    executor = ThreadPoolExecutor(max_workers=3)
    app["executor"] = executor
    return app


if __name__ == "__main__":
    app = create_app()
    web.run_app(app)