Starlette Gather Background Tasks

starlette_asyncio_gather_background_tasks.py

import asyncio

import httpx
import uvicorn
from starlette.applications import Starlette
from starlette.background import BackgroundTasks
from starlette.responses import JSONResponse
from starlette.routing import Route

client = httpx.AsyncClient()


async def some_request():
    r = await client.get("https://www.python-httpx.org/")
    return r


class GatherBackgroundTasks:
    def __init__(self, tasks: list | None = None):
        self.tasks = list(tasks) if tasks else []

    def add_task(self, func, *args, **kwargs) -> None:
        self.tasks.append(func(*args, **kwargs))

    async def __call__(self) -> None:
        await asyncio.gather(*self.tasks)


async def test_simple(*_, **__):
    tasks = BackgroundTasks()
    tasks.add_task(some_request)
    tasks.add_task(some_request)
    return JSONResponse({"hello": "world1"}, background=tasks)


async def test_gather(*_, **__):
    tasks = GatherBackgroundTasks()
    tasks.add_task(some_request)
    tasks.add_task(some_request)
    return JSONResponse({"hello": "world2"}, background=tasks)


routes = [
    Route("/simple/", endpoint=test_simple),
    Route("/gather/", endpoint=test_gather),
]

app = Starlette(debug=True, routes=routes)
if __name__ == "__main__":
    uvicorn.run(app)