FastAPI Request Background Tasks

Introduction

When building web applications, you'll often encounter situations where certain operations shouldn't block the main request flow. For example, you might want to send confirmation emails, process uploaded files, or update analytics data after responding to a user's request. This is where FastAPI's background tasks feature comes in handy.

Background tasks in FastAPI allow you to execute code after the response has been sent to the client, making your API more responsive while still performing necessary operations. In this tutorial, we'll dive into how to implement and manage these tasks effectively.

What Are Background Tasks?

Background tasks are operations that run "in the background" after the API has already sent a response to the client. They're useful when:

• You need to perform operations that don't affect the response
• You want to improve response time by deferring non-critical processing
• You want to handle tasks that might take a while to complete

FastAPI provides a simple way to create background tasks without needing to set up complex task queuing systems (although for more complex scenarios, you might still need solutions like Celery).

Basic Background Task Implementation

Let's start with a simple example:

python

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_notification(email: str, message: str):
    with open("notifications.txt", mode="a") as file:
        content = f"notification for {email}: {message}\n"
        file.write(content)

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="Hello World!")
    return {"message": "Notification sent in the background"}

In this example:

1. We define a simple function write_notification that writes a message to a file
2. We inject the BackgroundTasks object into our endpoint function
3. We add our task to the background tasks queue using add_task()
4. The API immediately returns a response, and our task runs afterward

The client receives a response quickly, and the file writing happens in the background after the response is sent.

Background Tasks with Dependencies

You can also use FastAPI's dependency injection system with background tasks. This is useful when your background tasks need access to database connections or other resources:

python

from fastapi import FastAPI, BackgroundTasks, Depends

app = FastAPI()

def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

def process_item(item_id: int, db: DBSession):
    # This runs in the background after the response is sent
    item = db.query(Item).filter(Item.id == item_id).first()
    # Process the item...
    db.commit()

@app.post("/items/{item_id}/process")
async def process_item_endpoint(
    item_id: int, 
    background_tasks: BackgroundTasks,
    db: DBSession = Depends(get_db)
):
    background_tasks.add_task(process_item, item_id, db)
    return {"message": "Item will be processed in the background"}

Handling Multiple Background Tasks

You can add multiple tasks to the same background tasks object:

python

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, body: str):
    # Logic to send an email
    print(f"Sending email to {email} with body: {body}")

def update_stats(user_id: int):
    # Logic to update user stats
    print(f"Updating stats for user {user_id}")

@app.post("/users/{user_id}/register")
async def register_user(
    user_id: int,
    email: str,
    background_tasks: BackgroundTasks
):
    # Add multiple tasks to be run in the background
    background_tasks.add_task(send_email, email, body="Welcome to our service!")
    background_tasks.add_task(update_stats, user_id)
    
    return {"message": "User registered successfully"}

The tasks will be executed in the order they were added after the response is sent.

Error Handling in Background Tasks

It's important to note that exceptions in background tasks won't affect the API response since the response is already sent. However, those exceptions will still be raised and might crash your application if not handled properly. Always include try-except blocks in your background task functions:

python

def process_data_safely(data: dict):
    try:
        # Potentially dangerous operations
        result = 10 / data.get("divisor", 0)  # Could cause ZeroDivisionError
        process_result(result)
    except Exception as e:
        # Log the error instead of crashing
        print(f"Background task error: {e}")
        # You might want to log to a file or monitoring service in production

@app.post("/process")
async def process_endpoint(data: dict, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_data_safely, data)
    return {"message": "Processing started"}

Real-World Example: Image Processing API

Let's build a practical example of an image processing API that immediately returns a response while processing the uploaded image in the background:

python

from fastapi import FastAPI, BackgroundTasks, UploadFile, File
from fastapi.responses import JSONResponse
import shutil
import os
from pathlib import Path
import time

app = FastAPI()

UPLOAD_DIR = Path("uploads")
PROCESSED_DIR = Path("processed")

# Ensure directories exist
UPLOAD_DIR.mkdir(exist_ok=True)
PROCESSED_DIR.mkdir(exist_ok=True)

def process_image(image_path: str):
    """
    Simulate image processing that takes time
    In a real app, this could be resizing, filtering, or other operations
    """
    time.sleep(5)  # Simulate long processing
    
    # Create a processed version (here we just copy it)
    filename = os.path.basename(image_path)
    processed_path = PROCESSED_DIR / filename
    
    try:
        # In a real app, you would apply image transformations here
        shutil.copy(image_path, processed_path)
        
        # You could update a database with the status
        print(f"Processed image: {filename}")
    except Exception as e:
        print(f"Error processing image {filename}: {e}")

@app.post("/upload-image/")
async def upload_image(
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...)
):
    # Save the uploaded file
    file_path = UPLOAD_DIR / file.filename
    
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    # Schedule the image processing in the background
    background_tasks.add_task(process_image, str(file_path))
    
    return JSONResponse(
        status_code=202,
        content={
            "message": "Image uploaded successfully and is being processed",
            "filename": file.filename
        }
    )

@app.get("/status/{filename}")
async def get_processing_status(filename: str):
    # In a real app, you would check a database for the status
    processed_file = PROCESSED_DIR / filename
    
    if processed_file.exists():
        return {"status": "completed", "filename": filename}
    elif (UPLOAD_DIR / filename).exists():
        return {"status": "processing", "filename": filename}
    else:
        return {"status": "not_found", "filename": filename}

In this example:

1. Users upload an image
2. The API immediately returns a 202 Accepted response
3. The image processing happens in the background
4. Users can check the processing status with a separate endpoint

When to Use Background Tasks vs. Task Queues

FastAPI's background tasks are great for simple, short-lived operations. However, they have limitations:

• They run in the same process as your API
• If your server restarts, incomplete tasks are lost
• They're not distributed across multiple workers

For more complex requirements, consider dedicated task queue systems like Celery, RQ, or using cloud solutions like AWS Lambda. FastAPI integrates well with these solutions.

Summary

FastAPI's background tasks feature provides a straightforward way to perform operations after sending a response to the client, making your API more responsive and user-friendly. Key takeaways:

• Background tasks execute after the response is sent
• They're perfect for non-critical operations that shouldn't block the response
• You can add multiple tasks to be executed in sequence
• Always handle exceptions in background tasks to prevent application crashes
• For complex or long-running tasks, consider dedicated task queuing systems

Additional Resources

• FastAPI Official Documentation on Background Tasks
• Advanced Concurrency with FastAPI
• Integrating Celery with FastAPI

Exercises

1. Create a FastAPI endpoint that records user login events in a background task
2. Build an API that sends confirmation emails in the background after user registration
3. Modify the image processing example to include multiple processing operations (resize, convert format, etc.)
4. Implement proper error handling and logging in a background task
5. Create a background task that depends on database access using FastAPI's dependency injection