Background tasks¶

Some kills should not wait for the feast. When a client needs a quick response but the server still has slow work to do, schedule that work to run after the response has been sent. VelociPy’s BackgroundTasks lets you register side effects without blocking the handler or keeping the request alive.

Background tasks are perfect for sending emails, writing audit logs, flushing analytics, or triggering downstream webhooks. The response returns first; the claws keep working behind the scenes.

Scheduling post-response work¶

Declare a BackgroundTasks parameter in your handler, then call add_task with the function and its arguments.

import asyncio
from typing import Any
from velocipy import BackgroundTasks, VelociPy

app = VelociPy()


async def send_welcome_email(email: str) -> None:
    await asyncio.sleep(0.1)  # Simulate an API call.
    print(f"Sent welcome email to {email}")


@app.post("/signup")
async def signup(
    data: dict[str, Any],
    background_tasks: BackgroundTasks,
) -> dict[str, str]:
    email = data.get("email", "[email protected]")
    background_tasks.add_task(send_welcome_email, email)
    return {"status": "ok", "email": email}

The client receives {"status": "ok", ...} immediately. The email sender runs once the response is in flight.

Tip

Task functions can be sync or async. VelociPy awaits async functions and calls sync functions directly.

Queuing multiple tasks¶

A single BackgroundTasks container can carry many tasks. They run in the order they were added.

async def log_event(email: str) -> None:
    print(f"Logged signup for {email}")


@app.post("/newsletter")
async def newsletter(
    data: dict[str, Any],
    background_tasks: BackgroundTasks,
) -> dict[str, str]:
    email = data["email"]
    background_tasks.add_task(log_event, email)
    background_tasks.add_task(send_welcome_email, email)
    return {"status": "subscribed"}

Passing extra arguments and keyword arguments¶

add_task forwards *args and **kwargs exactly like functools.partial.

async def notify_admin(event: str, *, priority: str = "normal") -> None:
    print(f"[{priority}] Admin notified: {event}")


@app.post("/report")
async def report(
    background_tasks: BackgroundTasks,
) -> dict[str, str]:
    background_tasks.add_task(
        notify_admin,
        "user_report",
        priority="high",
    )
    return {"status": "reported"}

Error handling¶

Background tasks run after the response has left the nest, so a failure cannot change the status code. VelociPy catches exceptions, logs them, and continues with the remaining tasks.

import logging

async def flaky_task() -> None:
    raise RuntimeError("Something went wrong in the underbrush")


@app.post("/fire-and-forget")
async def fire_and_forget(
    background_tasks: BackgroundTasks,
) -> dict[str, str]:
    background_tasks.add_task(flaky_task)
    return {"status": "accepted"}

Warning

Background tasks are fire-and-forget. If you need delivery guarantees, use a task queue such as Celery, RQ, or arq instead of in-process background tasks.

Advanced patterns¶

Using a dependency to prepare shared work¶

You can inject a configured helper via Depends, then add the actual task in the route.

from typing import Annotated
from velocipy import Depends


async def analytics():
    class Analytics:
        async def track(self, event: str) -> None:
            print(f"Tracked: {event}")
    return Analytics()


@app.post("/checkout")
async def checkout(
    background_tasks: BackgroundTasks,
    analytics: Annotated[Analytics, Depends(analytics)],
) -> dict[str, str]:
    background_tasks.add_task(analytics.track, "checkout_complete")
    return {"status": "confirmed"}

Note

BackgroundTasks itself cannot be declared inside dependencies. Schedule tasks only on the route’s own BackgroundTasks parameter.

Testing background tasks¶

Because tasks run after the response, normal TestClient calls still return synchronously. To verify the side effect, mock the task function and assert it was called, or test the task function separately.

from unittest.mock import MagicMock
from velocipy.testing import TestClient

app = VelociPy()

def test_signup_schedules_email():
    mock_send = MagicMock()
    app.dependency_overrides = {}

    @app.post("/signup-test")
    async def signup_test(background_tasks: BackgroundTasks):
        background_tasks.add_task(mock_send, "[email protected]")
        return {"status": "ok"}

    client = TestClient(app)
    response = client.post("/signup-test")
    assert response.status_code == 200
    mock_send.assert_called_once_with("[email protected]")

At a glance¶

Gear	What it does
`BackgroundTasks`	Container injected into a route to schedule deferred work.
`add_task(func, args, *kwargs)`	Schedules a sync or async function to run after the response.
Order	Tasks run in the order they are added.
Errors	Exceptions are logged; remaining tasks continue.
Scope	Only usable as a route parameter, not inside dependencies.

See examples/background_tasks.py for a runnable demo.