Using Background Tasks in FastAPI

FastAPI provides native support for background tasks through the BackgroundTasks class, enabling you to handle time-consuming operations efficiently without blocking your API's responsiveness.

---

What Are Background Tasks?

Background tasks allow you to defer the execution of certain operations until after your API response is returned. Common use cases include:

• Sending email notifications.
• Logging user actions.
• Updating a cache or database asynchronously.

---

Setting Up Your Environment

Prerequisites

Ensure you have:

• Python 3.7 or higher
• FastAPI and Uvicorn installed:

  pip install fastapi uvicorn
  

---

Example 1: Writing to a File in the Background

Step 1: Create a Task Function

Define a function to execute the background task:

def write_log(file_name: str, message: str):
    with open(file_name, "a") as file:
        file.write(f"{message}\n")

Step 2: Use BackgroundTasks in an Endpoint

Use the BackgroundTasks class to schedule the execution of the task after sending the response:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

@app.post("/log/")
async def log_message(message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "log.txt", message)
    return {"message": "Log task has been scheduled"}

How It Works: - The write_log function writes a message to a file. - The background_tasks.add_task method schedules the task to run after the response is sent.

---

Example 2: Sending Notifications

Step 1: Define the Task Function

Simulate sending an email or notification by writing to a log file:

def send_email(email: str, content: str):
    with open("emails.txt", "a") as email_file:
        email_file.write(f"Email to {email}: {content}\n")

Step 2: Use BackgroundTasks in the Endpoint

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "This is your notification content.")
    return {"message": f"Notification scheduled for {email}"}

Key Points: - This approach avoids blocking the endpoint while the email task runs. - Multiple tasks can be scheduled in a single request.

---

Example 3: Chaining Multiple Background Tasks

You can schedule multiple background tasks for a single request:

@app.post("/complex-task/")
async def complex_task(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "task_log.txt", "Task started")
    background_tasks.add_task(send_email, "[email protected]", "Your task is complete!")
    return {"message": "Complex task scheduled"}

How It Works: - Tasks are executed in the order they are added. - Tasks run asynchronously without delaying the response.

---

Advanced Use Cases

Passing Stateful Objects to Background Tasks

You can pass objects or configurations to tasks:

class Config:
    def __init__(self, service_name: str):
        self.service_name = service_name

def log_service_event(config: Config, event: str):
    with open("service_log.txt", "a") as log_file:
        log_file.write(f"[{config.service_name}] {event}\n")

@app.post("/service-event/")
async def service_event(event: str, background_tasks: BackgroundTasks):
    config = Config(service_name="MyService")
    background_tasks.add_task(log_service_event, config, event)
    return {"message": "Service event logged"}

---

Wrapping Up

FastAPI's BackgroundTasks is a powerful and simple way to execute tasks asynchronously without introducing external libraries. It’s best suited for lightweight and fast tasks like logging, notifications, or cache updates.

For more complex scenarios requiring task scheduling or distributed processing, consider integrating tools like Celery or RQ.

For production-ready setups and advanced features, check out Craft Your Startup — your go-to FastAPI boilerplate for building modern applications!