FastAPI in Practice: Add Background Jobs Without Blocking Requests

Pictalingo

·

FastAPI in Practice: Add Background Jobs Without Blocking Requests

FastAPI is often used for quick APIs: receive a request, validate data, return JSON. But real web apps usually need extra work after the response: sending emails, resizing images, calling a webhook, writing audit logs, or generating a report. If you do all of that inside the request handler, users wait longer and your API becomes fragile.

This article shows a practical pattern for junior and mid-level developers: use FastAPI’s built-in BackgroundTasks for simple background work, then structure the code so you can later move the job to a real queue like Celery, RQ, or Dramatiq.

When BackgroundTasks Is a Good Fit

BackgroundTasks runs work after FastAPI sends the response. It is useful for small, non-critical tasks that should happen soon but do not need a separate worker process.

  • Good fit: sending a welcome email after signup.
  • Good fit: writing a small audit log.
  • Good fit: notifying an internal webhook.
  • Bad fit: processing a 500 MB video.
  • Bad fit: long-running reports that must survive server restarts.
  • Bad fit: jobs requiring retries, scheduling, or distributed workers.

The important rule is simple: background tasks are convenient, but they are not a durable queue. If the server crashes before the task finishes, the task may be lost.

Project Setup

Create a small FastAPI project:

mkdir fastapi-background-jobs cd fastapi-background-jobs python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn pydantic

Use this structure:

fastapi-background-jobs/ app/ main.py schemas.py services.py tasks.py

This keeps HTTP logic separate from business logic. That separation matters because background jobs should not be tightly coupled to request handlers.

Define the Request Schema

In app/schemas.py, create a simple model for user registration:

from pydantic import BaseModel, EmailStr class UserCreate(BaseModel): email: EmailStr full_name: str class UserResponse(BaseModel): id: int email: EmailStr full_name: str message: str

You will also need the email validator dependency:

pip install "pydantic[email]"

The schema gives you clean validation before your route logic runs. If the email is invalid, FastAPI returns a validation error automatically.

Create a Fake User Service

In a real app, this would use a database. For this tutorial, use an in-memory list so the example stays focused.

Create app/services.py:

from app.schemas import UserCreate _USERS = [] _NEXT_ID = 1 def create_user(data: UserCreate) -> dict: global _NEXT_ID user = { "id": _NEXT_ID, "email": data.email, "full_name": data.full_name, } _USERS.append(user) _NEXT_ID += 1 return user def list_users() -> list[dict]: return _USERS

This service does not know anything about HTTP, routes, or background tasks. It only creates and returns users. That makes it easy to test and easy to reuse.

Add Background Task Functions

Create app/tasks.py:

from datetime import datetime from pathlib import Path import time LOG_FILE = Path("audit.log") def send_welcome_email(email: str, full_name: str) -> None: """ Simulates sending an email. In production, replace this with a real email provider call, such as Amazon SES, Mailgun, Postmark, or SendGrid. """ time.sleep(2) print(f"Welcome email sent to {full_name} <{email}>") def write_audit_log(action: str, user_email: str) -> None: timestamp = datetime.utcnow().isoformat() with LOG_FILE.open("a", encoding="utf-8") as file: file.write(f"{timestamp} | {action} | {user_email}\n")

These functions are plain Python functions. That is intentional. Do not put FastAPI request objects inside your tasks. Pass only the data the task needs, such as email, full_name, or user_id.

Wire It Into FastAPI

Create app/main.py:

from fastapi import BackgroundTasks, FastAPI from app.schemas import UserCreate, UserResponse from app.services import create_user, list_users from app.tasks import send_welcome_email, write_audit_log app = FastAPI(title="FastAPI Background Jobs Demo") @app.post("/users", response_model=UserResponse, status_code=201) def register_user( payload: UserCreate, background_tasks: BackgroundTasks, ): user = create_user(payload) background_tasks.add_task( send_welcome_email, user["email"], user["full_name"], ) background_tasks.add_task( write_audit_log, "user_registered", user["email"], ) return { **user, "message": "User created. Background tasks scheduled.", } @app.get("/users") def get_users(): return {"users": list_users()}

Run the app:

uvicorn app.main:app --reload

Test it with curl:

curl -X POST http://127.0.0.1:8000/users \ -H "Content-Type: application/json" \ -d '{"email":"[email protected]","full_name":"Maya Chen"}'

The API responds immediately with something like this:

{ "id": 1, "email": "[email protected]", "full_name": "Maya Chen", "message": "User created. Background tasks scheduled." }

About two seconds later, you should see the email simulation printed in the terminal. You should also see an audit.log file created in the project root.

Handle Errors Inside Tasks

A common mistake is assuming background task failures behave like route failures. They do not. The response has already been sent, so you cannot return a normal API error to the client.

Wrap risky task logic with logging:

import logging import time logger = logging.getLogger(__name__) def send_welcome_email(email: str, full_name: str) -> None: try: time.sleep(2) # Replace this with a real provider API call. print(f"Welcome email sent to {full_name} <{email}>") except Exception: logger.exception("Failed to send welcome email to %s", email)

In production, configure structured logs and send them to your monitoring system. At minimum, make sure background task errors are visible. Silent failures are painful to debug.

A Cleaner Pattern: Task Wrapper Functions

As your app grows, route handlers can become messy if they schedule many tasks directly. A cleaner option is to create a task orchestration function.

Update app/tasks.py:

from fastapi import BackgroundTasks def schedule_user_created_tasks( background_tasks: BackgroundTasks, email: str, full_name: str, ) -> None: background_tasks.add_task(send_welcome_email, email, full_name) background_tasks.add_task(write_audit_log, "user_registered", email)

Then simplify the route:

from fastapi import BackgroundTasks, FastAPI from app.schemas import UserCreate, UserResponse from app.services import create_user, list_users from app.tasks import schedule_user_created_tasks app = FastAPI(title="FastAPI Background Jobs Demo") @app.post("/users", response_model=UserResponse, status_code=201) def register_user( payload: UserCreate, background_tasks: BackgroundTasks, ): user = create_user(payload) schedule_user_created_tasks( background_tasks=background_tasks, email=user["email"], full_name=user["full_name"], ) return { **user, "message": "User created. Background tasks scheduled.", }

The route now reads like a normal request flow: validate input, create user, schedule side effects, return response.

Testing the Route

You can test the API with FastAPI’s TestClient. Install pytest first:

pip install pytest httpx

Create test_users.py:

from fastapi.testclient import TestClient from app.main import app client = TestClient(app) def test_register_user(): response = client.post( "/users", json={ "email": "[email protected]", "full_name": "Sam Rivera", }, ) assert response.status_code == 201 data = response.json() assert data["email"] == "[email protected]" assert data["full_name"] == "Sam Rivera" assert data["message"] == "User created. Background tasks scheduled."

Run the test:

pytest

For larger applications, you may want to mock email providers and file writes. The main thing to test at the route level is that the user receives the correct response. Test task behavior separately as plain Python functions.

When to Move to a Real Queue

FastAPI background tasks are a good first step, but they are not the final architecture for every workload. Move to a real queue when you need stronger guarantees.

  • You need retries when an email or webhook fails.
  • You need job status, such as pending, running, or failed.
  • You need scheduled jobs or delayed execution.
  • Your tasks are CPU-heavy or take more than a few seconds.
  • You run multiple API servers and need centralized job processing.

A good migration path is to keep task functions independent from FastAPI. For example, this function can later be called by Celery without changing your route’s business logic:

def send_welcome_email(email: str, full_name: str) -> None: # Provider call goes here. ...

That small design choice saves time later. Your API layer schedules work, but the task function owns the work.

Practical Checklist

  • Use BackgroundTasks only for short, non-critical side effects.
  • Keep task functions as plain Python functions.
  • Pass simple data into tasks, not request objects or database sessions.
  • Log exceptions inside background tasks.
  • Move to Celery, RQ, or another queue when you need retries or durability.

Conclusion

FastAPI’s BackgroundTasks gives you a simple way to keep API responses fast while still performing useful side effects. The key is to use it deliberately. Keep the route clean, isolate task logic, log failures, and know when the workload has outgrown the built-in tool.

For many small and medium web apps, this pattern is enough to improve user experience without adding Redis, workers, or queue infrastructure on day one.

Leave a Reply

Your email address will not be published. Required fields are marked *