How to Use BackgroundTasks in Air for Asynchronous Processing

Use Air's BackgroundTasks class to offload work from HTTP responses by injecting it as a dependency or instantiating it manually, scheduling functions with add_task() to run after the response is sent.

Air provides a lightweight, type-safe wrapper around FastAPI's background task infrastructure, enabling you to perform fire-and-forget operations without blocking API clients. Located in src/air/background.py, this utility inherits from FastAPI's BackgroundTasks while adding enhanced type hints and documentation, ensuring that any function scheduled via add_task() executes only after the HTTP response has been transmitted.

Understanding the BackgroundTasks Implementation

The BackgroundTasks class in src/air/background.py extends FastAPIBackgroundTasks without altering the underlying execution model. Air adds a generic add_task[**P, T] method that accepts a callable function and its arguments, storing them for post-response execution.

# src/air/background.py

class BackgroundTasks(FastAPIBackgroundTasks):
    def add_task[**P, T](
        self,
        func: Annotated[Callable[P, T], Doc("Function to call after response sent")],
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> None:
        return super().add_task(func, *args, **kwargs)

FastAPI (via Starlette) manages the actual execution using a thread pool for synchronous functions and the event loop for coroutines. This architecture ensures that adding a background task introduces no latency to the HTTP response itself.

Adding Background Tasks in Request Handlers

To schedule work within an endpoint, declare air.BackgroundTasks as a dependency parameter. Air injects an instance tied to the current request context, allowing you to register tasks using add_task().

import pathlib
import air

app = air.Air()

def write_notification(email: str, message: str = "") -> None:
    content = f"notification for {email}: {message}"
    pathlib.Path("log.txt").write_text(content)

@app.post("/send-notification/{email}")
def send_notification(
    email: str, background_tasks: air.BackgroundTasks
) -> air.P:
    """Schedule file write without blocking the client."""
    background_tasks.add_task(write_notification, email, message="signup complete")
    return air.P(f"Notification queued for {email}")

As shown in examples/src/background__BackgroundTasks.py, the client receives an immediate response while the file system operation continues in the background. The method accepts both positional and keyword arguments, forwarding them directly to the target function.

Using BackgroundTasks Outside of Requests

You can instantiate BackgroundTasks directly for use in scripts, tests, or manual execution workflows. This pattern is demonstrated in tests/test_background.py, where tasks are created and invoked outside the standard request lifecycle.

import air

def heavy_computation(x: int) -> int:
    return x * x

# Manual instantiation

tasks = air.BackgroundTasks()
tasks.add_task(heavy_computation, 42)

# Execute immediately for testing

tasks.tasks[0].func()  # Returns 1764

When instantiated manually, you assume responsibility for triggering execution, as the automatic post-response hook only exists within the FastAPI request handling context.

Handling Async Functions in Background Tasks

The add_task() method detects coroutine functions automatically and schedules them appropriately within the asyncio event loop. No manual asyncio.create_task() wrapping is required.

import asyncio
import air

async def send_email(recipient: str) -> None:
    await asyncio.sleep(1)  # Simulates network I/O

    print(f"Email sent to {recipient}")

@app.get("/welcome/{addr}")
def welcome_user(addr: str, background_tasks: air.BackgroundTasks):
    background_tasks.add_task(send_email, addr)
    return air.P("Welcome email scheduled")

FastAPI's background runner distinguishes between synchronous callables (executed in a thread pool) and asynchronous callables (awaited in the event loop), ensuring optimal resource utilization regardless of function type.

Summary

• Air's BackgroundTasks in src/air/background.py provides a typed wrapper around FastAPI's background execution system.
• Dependency injection via background_tasks: air.BackgroundTasks automatically handles post-response execution within HTTP endpoints.
• Manual instantiation allows use in scripts and tests, though execution must be triggered manually in these contexts.
• Universal support for both def and async def functions eliminates the need for manual async wrapping or thread management.
• Zero response latency because task registration occurs immediately while execution happens only after the HTTP response is transmitted.

Frequently Asked Questions

What is the difference between Air's BackgroundTasks and FastAPI's?

Air's BackgroundTasks inherits directly from FastAPIBackgroundTasks and maintains identical runtime behavior. The Air version adds enhanced type hints using Python generics and comprehensive docstrings, providing better IDE support and type checking without changing the underlying execution model implemented in FastAPI and Starlette.

Can BackgroundTasks fail silently if the server crashes?

Background tasks run after the HTTP response is sent, making them vulnerable to interruption if the server process terminates or crashes immediately following a request. For critical operations requiring durability guarantees, implement a proper task queue using Redis, Celery, or RQ rather than relying on in-process background tasks.

How do I pass complex objects to background tasks?

The add_task() method accepts any combination of positional (*args) and keyword (**kwargs) arguments that the target function supports. However, avoid passing database connections or file handles, as these may close before the background task executes. Instead, pass identifiers or primitive data types that allow the task to reconstruct necessary resources independently.

Is there a limit to how many tasks I can add per request?

While Air and FastAPI impose no explicit limit on task registration, each background task consumes a thread from Starlette's thread pool (for sync functions) or requires event loop time (for async functions). Excessive queuing can exhaust thread pools or memory; for high-volume processing, offload work to external task queues rather than queuing hundreds of tasks within a single HTTP response cycle.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}