Demonstration of using a thread pool to launch background tasks in a web server

README.md

Start the Server

python toy.py

Create a Job

curl -X PUT localhost:8080/[name]

Job Status

curl localhost:8080/[name]

toy.py

import subprocess
from concurrent.futures import ThreadPoolExecutor
from time import sleep

from aiohttp import web


def job_run(name):
    if name.startswith("x"):
        raise Exception("I don't like jobs that start with 'x'")

    sleep(10)
    return subprocess.run("cat", text=True, input=f"Hello from {name}", capture_output=True)

async def job_start(request):
    jobs = request.app["jobs"]
    pool = request.app["pool"]

    job_id = request.match_info["job"]
    if job_id not in jobs:
        jobs[job_id] = pool.submit(job_run, job_id)

    return await job_status(request)

async def job_status(request):
    try:
        job_id = request.match_info["job"]
        job = request.app["jobs"][job_id]

    except KeyError:
        raise web.HTTPNotFound(text=f"No such job: {job_id}")

    if job.running():
        return web.json_response({
            "status": "pending"
        })

    if job.done():
        if (exc := job.exception()) is not None:
            return web.json_response({
                "status": "failed",
                "output": str(exc)
            })

        return web.json_response({
            "status": "done",
            "output": job.result().stdout
        })

    return web.json_response({
        "status": "submitted"
    })

async def shutdown(app):
    app["pool"].shutdown()

if __name__ == "__main__":
    app = web.Application()
    app.on_shutdown.append(shutdown)

    app["jobs"] = {}
    app["pool"] = ThreadPoolExecutor()

    app.router.add_get("/{job}", job_status)
    app.router.add_put("/{job}", job_start)

    web.run_app(app)