Build Robust APIs: Dependencies, Errors & Background Tasks

What You’ll Learn

In this chapter, you’ll gain essential skills to build more robust and maintainable FastAPI applications. You will learn:

How to leverage FastAPI’s Dependency Injection system to write reusable logic and manage shared resources.
Practical applications of dependencies, such as handling database sessions and conceptually implementing authentication.
How to effectively use Custom HTTP Exceptions to provide clear and graceful error messages in your API.
How to implement Background Tasks for performing non-blocking operations after a response has been sent.

Core Concepts

Dependency Injection: Reusable Logic and Shared Resources

Imagine you have several API endpoints that all need to perform the same initial check, like validating an API key, getting a database connection, or fetching a specific user from a database. Copying and pasting this code into every endpoint would be tedious and error-prone. This is where Dependency Injection (DI) comes in.

In FastAPI, DI is a powerful feature that allows your path operations (your API endpoints) to declare “dependencies” – things they need to function. FastAPI then automatically provides these dependencies when the path operation is called. This promotes code reuse, makes your code easier to test, and keeps your path operations focused on their main task.

You declare a dependency by using the fastapi.Depends() function.

from fastapi import FastAPI, Depends, HTTPException, status

app = FastAPI()

# A simple dependency function
def get_current_user(token: str):
    if token != "mysecrettoken":
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    return {"username": "admin", "id": 1}

@app.get("/items/")
async def read_items(user: dict = Depends(get_current_user)):
    return {"message": f"Hello, {user['username']}! Here are your items."}

In this example, read_items depends on get_current_user. Before read_items runs, FastAPI calls get_current_user, passing the token (which it expects from a query parameter or header). If get_current_user raises an exception, read_items won’t even be called. If it returns a value, that value is passed to the user parameter of read_items.

Common Use Cases for Dependencies: Database Sessions

Database connections are a classic example where dependencies shine. You often need to:

Open a database connection/session.
Use it within your path operation.
Close the connection/session, regardless of whether an error occurred.

FastAPI dependencies can use yield to handle setup and teardown logic. The code before yield runs when the dependency is called, and the code after yield runs after the path operation has finished (or an error occurred).

# This is a simplified example, a real database connection would be more complex
def get_database_session():
    print("Opening database session...")
    db_session = {"connection": "active", "data": []} # Simulate a session
    try:
        yield db_session # Provide the session to the path operation
    finally:
        print("Closing database session...")
        db_session["connection"] = "closed" # Clean up

When a path operation depends on get_database_session, it receives db_session. After the path operation completes, the finally block in get_database_session executes, ensuring cleanup.

Authentication (Conceptual)

Dependencies are the perfect place to implement authentication and authorization logic. While we won’t build a full authentication system here, you can see how a dependency can protect an endpoint.

# Conceptual Authentication Dependency
def requires_admin_privileges(user: dict = Depends(get_current_user)):
    if user["username"] != "admin":
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough privileges")
    return user

@app.get("/admin_dashboard/")
async def admin_dashboard(admin_user: dict = Depends(requires_admin_privileges)):
    return {"message": f"Welcome to the admin dashboard, {admin_user['username']}!"}

Here, requires_admin_privileges itself depends on get_current_user. This creates a chain of dependencies, where get_current_user is called first, then its result is passed to requires_admin_privileges, and finally, the result of that is passed to admin_dashboard.

Custom HTTP Exceptions: Handling API Errors Gracefully

When something goes wrong in your API, you shouldn’t just let Python’s default error messages be returned. These are often unhelpful to API consumers and might expose internal details. FastAPI provides HTTPException to raise standard HTTP errors with custom messages.

You raise an HTTPException with a status_code and a detail message. FastAPI automatically converts this into a JSON response that follows the OpenAPI standard, making it easy for client applications to understand.

from fastapi import HTTPException, status

@app.get("/items/{item_id}")
async def read_single_item(item_id: int):
    if item_id not in [1, 2, 3]: # Simulate item lookup
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item with ID {item_id} not found."
        )
    return {"item_id": item_id, "name": f"Item {item_id}"}

When item_id is not 1, 2, or 3, the API will return a 404 Not Found status with a clear JSON message: {"detail": "Item with ID X not found."}.

Background Tasks

Sometimes you have operations that are important but don’t need to block the user’s request. For example, sending a welcome email after a user signs up, logging an event, or processing an image. FastAPI’s BackgroundTasks feature allows you to run functions after the HTTP response has been sent to the client.

This means the user gets a fast response, and the background task runs asynchronously.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    with open("api_log.txt", "a") as log_file:
        log_file.write(message + "\n")

@app.post("/send-message/")
async def send_message(message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Received message: {message}")
    return {"status": "Message received, processing in background."}

When a request hits /send-message/, the API immediately returns the success message. Then, write_log is called in the background, logging the message without delaying the client’s response.

Hands-On Practice

Let’s get your hands dirty with some coding! For each exercise, create a new Python file (e.g., chapter_dependencies.py) and run it using uvicorn chapter_dependencies:app --reload. Test your endpoints using your browser or a tool like curl or Postman.

Exercise 1: Simple Dependency for Query Parameter Validation

Goal: Create a dependency that ensures a query parameter has a minimum length.

Create the Dependency: Write a function validate_search_query that takes a query: str as an argument. If len(query) is less than 3, raise an HTTPException with status_code=400 and detail="Query must be at least 3 characters long.". Otherwise, return the query.
Use the Dependency: Create a path operation /search/ that accepts a query: str as a query parameter and uses validate_search_query as a dependency. The path operation should return a message like {"search_results": f"Searching for: {query}"}.

# chapter_dependencies.py
from fastapi import FastAPI, Depends, HTTPException, status

app = FastAPI()

def validate_search_query(query: str):
    if len(query) < 3:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Query must be at least 3 characters long."
        )
    return query

@app.get("/search/")
async def search_items(valid_query: str = Depends(validate_search_query)):
    return {"search_results": f"Searching for: {valid_query}"}

Expected Outcome:

Access /search/?query=abc: Returns {"search_results": "Searching for: abc"}.
Access /search/?query=ab: Returns {"detail": "Query must be at least 3 characters long."} with a 400 status code.

Debugging Tips:

Check the HTTP status code in your browser’s developer tools or curl -v.
Ensure your Depends() call is correct.

Exercise 2: Simulating a Database Session Dependency

Goal: Create a yield-based dependency that simulates opening and closing a database session.

Create the get_db Dependency: Write an async function get_db. Inside, print “Opening mock DB session…”, then yield a string like "MockDBConnection_123". In a finally block, print “Closing mock DB session…”.
Use the Dependency: Create a path operation /data/ that takes db_session: str = Depends(get_db). It should return {"message": "Data retrieved using", "session": db_session}.

# chapter_dependencies.py (add to existing file)
# ... imports and app = FastAPI() ...

async def get_db():
    print("Opening mock DB session...")
    mock_db_session = "MockDBConnection_123"
    try:
        yield mock_db_session
    finally:
        print("Closing mock DB session...")

@app.get("/data/")
async def get_some_data(db_session: str = Depends(get_db)):
    return {"message": "Data retrieved using", "session": db_session}

Expected Outcome:

Access /data/: Returns {"message": "Data retrieved using", "session": "MockDBConnection_123"}.
In your terminal where uvicorn is running, you should see “Opening mock DB session…” before the API response, and “Closing mock DB session…” after the response.

Debugging Tips:

If you don’t see the print statements, ensure your uvicorn server is running in the terminal you’re watching.
Double-check async and await keywords if you encounter runtime errors related to coroutines.

Exercise 3: Custom HTTP Exception for Item Not Found

Goal: Implement a custom HTTPException when an item is not found.

Create Mock Data: Define a simple dictionary fake_items_db = {1: {"name": "Laptop"}, 2: {"name": "Mouse"}}.
Implement Path Operation: Create a path operation /items/{item_id} that takes item_id: int. If item_id is not in fake_items_db, raise an HTTPException with status_code=404 and detail=f"Item {item_id} not found.". Otherwise, return the item from fake_items_db.

# chapter_dependencies.py (add to existing file)
# ... imports and app = FastAPI() ...

fake_items_db = {
    1: {"name": "Laptop"},
    2: {"name": "Mouse"},
    3: {"name": "Keyboard"}
}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in fake_items_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {item_id} not found."
        )
    return fake_items_db[item_id]

Expected Outcome:

Access /items/1: Returns {"name": "Laptop"}.
Access /items/99: Returns {"detail": "Item 99 not found."} with a 404 status code.

Debugging Tips:

Verify the status_code and detail message in the API response.
Ensure your item_id type hint is int.

Exercise 4: Conceptual Authentication Dependency

Goal: Protect an endpoint using a dependency that checks for an API key in a header.

Create Authentication Dependency: Write a function verify_api_key that takes x_api_key: str | None = Header(default=None) as an argument. If x_api_key is None or not equal to "SUPER_SECRET_KEY", raise an HTTPException with status_code=401 and detail="Unauthorized: Invalid API Key". Return True if the key is valid.
Protect an Endpoint: Create a path operation /protected_info/ that uses verify_api_key as a dependency. It should return {"message": "Welcome, authorized user!"}.

# chapter_dependencies.py (add to existing file)
# ... imports and app = FastAPI() ...
from fastapi import Header

def verify_api_key(x_api_key: str | None = Header(default=None)):
    if x_api_key is None or x_api_key != "SUPER_SECRET_KEY":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Unauthorized: Invalid API Key"
        )
    return True

@app.get("/protected_info/")
async def get_protected_info(authenticated: bool = Depends(verify_api_key)):
    return {"message": "Welcome, authorized user!"}

Expected Outcome:

Access /protected_info/ without X-API-Key header: Returns {"detail": "Unauthorized: Invalid API Key"} with a 401 status.
Access /protected_info/ with X-API-Key: WRONG_KEY: Returns {"detail": "Unauthorized: Invalid API Key"} with a 401 status.
Access /protected_info/ with X-API-Key: SUPER_SECRET_KEY: Returns {"message": "Welcome, authorized user!"}.

Debugging Tips:

Use curl -H "X-API-Key: SUPER_SECRET_KEY" http://localhost:8000/protected_info/ to test with headers.
Ensure Header(default=None) is correctly imported and used.

Exercise 5: Background Task for Logging

Goal: Implement a background task to log a message after the API response is sent.

Create Background Function: Write a function log_activity(activity_message: str) that simply prints the activity_message to the console.
Implement Path Operation with Background Task: Create a path operation /perform_action/ that accepts a message: str as a query parameter. It should take background_tasks: BackgroundTasks as a dependency. Add log_activity to background_tasks with the received message. Return {"status": "Action initiated, logging in background."}.

# chapter_dependencies.py (add to existing file)
# ... imports and app = FastAPI() ...
from fastapi import BackgroundTasks

def log_activity(activity_message: str):
    print(f"BACKGROUND LOG: {activity_message}")

@app.post("/perform_action/")
async def perform_action(message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(log_activity, f"User performed action with message: '{message}'")
    return {"status": "Action initiated, logging in background."}

Expected Outcome:

Send a POST request to /perform_action/?message=hello_world.
You will receive an immediate {"status": "Action initiated, logging in background."} response.
Shortly after, in your terminal where uvicorn is running, you should see BACKGROUND LOG: User performed action with message: 'hello_world'.

Debugging Tips:

Ensure you are using a POST request.
Check your terminal output carefully for the log message.
If the background task isn’t running, verify background_tasks.add_task() is called correctly.

Real-World Application

Let’s combine these concepts into small, practical projects.

Project 1: A “Protected” Item Store

Build a simple API that manages a list of items. Some items are public, while others require authentication.

Setup:
- Start with app = FastAPI().
- Define fake_store_db = {"public_item": "Available to all", "private_item": "Only for admins"}.
Authentication Dependency:
- Re-use or adapt the verify_api_key dependency from Exercise 4 (e.g., check for X-ADMIN-Key header with value "ADMIN_SECRET"). Raise a 401 HTTPException if invalid.
Endpoints:
- GET /store/public/{item_name}: Returns the item from fake_store_db if item_name exists. If not, raise a 404 HTTPException. This endpoint is accessible to anyone.
- GET /store/private/{item_name}: This endpoint should depend on your admin authentication dependency. It behaves like the public endpoint but requires the X-ADMIN-Key header.

# project_store.py
from fastapi import FastAPI, Depends, HTTPException, status, Header

app = FastAPI()

fake_store_db = {
    "public_item": "Available to all",
    "private_item": "Only for admins"
}

def verify_admin_key(x_admin_key: str | None = Header(default=None)):
    if x_admin_key is None or x_admin_key != "ADMIN_SECRET":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Unauthorized: Admin key required"
        )
    return True

@app.get("/store/public/{item_name}")
async def get_public_item(item_name: str):
    if item_name not in fake_store_db:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Item '{item_name}' not found.")
    return {"item": fake_store_db[item_name]}

@app.get("/store/private/{item_name}")
async def get_private_item(item_name: str, admin_authenticated: bool = Depends(verify_admin_key)):
    if item_name not in fake_store_db:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Item '{item_name}' not found.")
    return {"item": fake_store_db[item_name]}

Project 2: User Activity Logger

Create an API where actions trigger a background log of user activity.

Setup:
- Start with app = FastAPI().
Background Logger Function:
- Create a function write_user_log(user_id: int, action: str) that prints a formatted log message to the console, including a timestamp. E.g., [YYYY-MM-DD HH:MM:SS] User {user_id} performed: {action}.
Endpoints:
- POST /users/{user_id}/track_activity: This endpoint accepts user_id: int and action: str (from query parameter or request body). It should immediately return {"status": "Activity logged in background."}.
- Add write_user_log to background_tasks with the user_id and action.

# project_logger.py
from fastapi import FastAPI, BackgroundTasks, Query
from datetime import datetime

app = FastAPI()

def write_user_log(user_id: int, action: str):
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[{timestamp}] User {user_id} performed: {action}")

@app.post("/users/{user_id}/track_activity")
async def track_user_activity(
    user_id: int,
    action: str = Query(..., description="The activity performed by the user"),
    background_tasks: BackgroundTasks = BackgroundTasks()
):
    background_tasks.add_task(write_user_log, user_id, action)
    return {"status": "Activity logged in background."}

Key Takeaways

Dependency Injection in FastAPI is a powerful pattern for creating reusable, testable, and modular API logic. Use Depends() to inject functions or classes.
Dependencies using yield are perfect for managing resources that require both setup and teardown, like database sessions.
Custom HTTP Exceptions (HTTPException) allow you to return standardized, clear, and informative error messages to API consumers, improving the user experience and debuggability.
Background Tasks enable your API to respond quickly to clients while performing less time-sensitive operations asynchronously, making your API more responsive and efficient.
These concepts are fundamental to building robust, scalable, and maintainable FastAPI applications in the real world.

Before moving to the next chapter, ensure you can comfortably:

Create and use simple dependencies.
Implement yield-based dependencies for resource management.
Raise HTTPException with appropriate status codes and details.
Add and observe the execution of background tasks.

Quick Reference

Here’s a cheat sheet for the key concepts covered in this chapter:

Dependency Injection:

from fastapi import Depends

# Basic Dependency
def get_user_id(user_id: int):
    # Perform validation or lookup
    return user_id

@app.get("/profile/")
async def get_profile(user: int = Depends(get_user_id)):
    return {"user_id": user}

# Yield-based Dependency (for setup/teardown)
async def get_db_session():
    db = "connect_to_db()" # Simulate connection
    try:
        yield db # Provide resource
    finally:
        "db.close()" # Clean up

@app.get("/data_from_db/")
async def get_data(session: str = Depends(get_db_session)):
    return {"data": f"Using {session}"}

Custom HTTP Exceptions:

from fastapi import HTTPException, status

# Raising an exception
@app.get("/product/{product_id}")
async def get_product(product_id: int):
    if product_id not in [1, 2]:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Product with ID {product_id} not found."
        )
    return {"product_id": product_id, "name": "Example Product"}

# Common Status Codes (from fastapi.status)
# status.HTTP_200_OK
# status.HTTP_201_CREATED
# status.HTTP_400_BAD_REQUEST (Bad Request - client error)
# status.HTTP_401_UNAUTHORIZED (Unauthorized - authentication missing/invalid)
# status.HTTP_403_FORBIDDEN (Forbidden - authenticated but no permission)
# status.HTTP_404_NOT_FOUND (Not Found - resource doesn't exist)
# status.HTTP_422_UNPROCESSABLE_ENTITY (Validation Error - pydantic errors)
# status.HTTP_500_INTERNAL_SERVER_ERROR (Internal Server Error)

Background Tasks:

from fastapi import BackgroundTasks

def send_email_in_background(recipient: str, subject: str, body: str):
    # Simulate sending email
    print(f"Sending email to {recipient} with subject '{subject}'")

@app.post("/signup/")
async def signup_user(email: str, background_tasks: BackgroundTasks):
    # User registration logic here
    background_tasks.add_task(send_email_in_background, email, "Welcome!", "Thanks for signing up!")
    return {"message": "User signed up, welcome email will be sent shortly."}

Build Robust APIs: Dependencies, Errors & Background Tasks

// table of contents

Build Robust APIs: Dependencies, Errors & Background Tasks

What You’ll Learn

Core Concepts

Dependency Injection: Reusable Logic and Shared Resources

Common Use Cases for Dependencies: Database Sessions

Authentication (Conceptual)

Custom HTTP Exceptions: Handling API Errors Gracefully

Background Tasks

Hands-On Practice

Exercise 1: Simple Dependency for Query Parameter Validation

Exercise 2: Simulating a Database Session Dependency

Exercise 3: Custom HTTP Exception for Item Not Found

Exercise 4: Conceptual Authentication Dependency

Exercise 5: Background Task for Logging

Real-World Application

Project 1: A “Protected” Item Store

Project 2: User Activity Logger

Key Takeaways

Quick Reference