‹ FastAPI Interview ☰
FastAPI Basics FastAPI Uvicorn FastAPI Routing FastAPI Pydantic Models FastAPI Automatic API Documentation FastAPI Handling Request Data FastAPI SQLAlchemy ORM FastAPI Tortoise ORM FastAPI Dependency Injection FastAPI Response Models FastAPI Error Handling FastAPI OAuth2 FastAPI JWT (Json Web Tokens) FastAPI RBAC FastAPI Background Tasks FastAPI Middleware FastAPI Handling File Uploads FastAPI Async Programming FastAPI WebSockets FastAPI Testing FastAPI Security FastAPI Extensions FastAPI CORS FastAPI Deployment FastAPI Optimization FastAPI GraphQL FastAPI Task Queues And Celery FastAPI Custom CLI Commands FastAPI Server-Sent Events

‹ FastAPI Interview

FastAPI Basics FastAPI Uvicorn FastAPI Routing FastAPI Pydantic Models FastAPI Automatic API Documentation FastAPI Handling Request Data FastAPI SQLAlchemy ORM FastAPI Tortoise ORM FastAPI Dependency Injection FastAPI Response Models FastAPI Error Handling FastAPI OAuth2 FastAPI JWT (Json Web Tokens) FastAPI RBAC FastAPI Background Tasks FastAPI Middleware FastAPI Handling File Uploads FastAPI Async Programming FastAPI WebSockets FastAPI Testing FastAPI Security FastAPI Extensions FastAPI CORS FastAPI Deployment FastAPI Optimization FastAPI GraphQL FastAPI Task Queues And Celery FastAPI Custom CLI Commands FastAPI Server-Sent Events

FastAPI Background Tasks

What are background tasks in FastAPI?

Background tasks in FastAPI allow you to execute functions outside of the main request-response cycle, meaning that tasks can run after the response is sent to the client. This is useful for tasks like sending emails, processing data, or updating logs without blocking the user from receiving a response promptly.

How do you use background tasks in FastAPI?

FastAPI provides a built-in class called BackgroundTasks to handle background execution. You add tasks to the background task queue and define the task function to be executed after the response is sent.

Example of using background tasks:

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(email: str, message: str):
    # Simulate sending email
    print(f"Sending email to {email}: {message}")

@app.post("/send-email/")
async def email_endpoint(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, "Welcome to our service!")
    return {"message": "Email sent in the background"}

In this example, the email is sent in the background using the send_email function. The client receives a response immediately while the email task is processed after the response.

How do background tasks work in FastAPI?

In FastAPI, background tasks work by adding the task to the internal background task queue using the BackgroundTasks class. Once the response is sent to the client, FastAPI continues executing the tasks in the background without affecting the user's experience.

The add_task() method is used to queue the task, and you can pass additional arguments to the task function as needed. The task is executed asynchronously, meaning it does not block the response.

How do you pass arguments to background tasks in FastAPI?

When adding a background task, you can pass any number of arguments to the task function. The arguments are passed as positional or keyword arguments when calling background_tasks.add_task().

Example of passing arguments to background tasks:

def save_user_data(user_id: int, data: dict):
    # Simulate saving data to a database
    print(f"Saving data for user {user_id}: {data}")

@app.post("/save-data/")
async def save_data(user_id: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(save_user_data, user_id, {"key": "value"})
    return {"message": "Data will be saved in the background"}

In this example, the save_user_data function is queued as a background task with the user ID and data passed as arguments.

How do you use multiple background tasks in FastAPI?

You can queue multiple background tasks in FastAPI by adding them to the BackgroundTasks object. Each task is executed sequentially after the response is sent to the client.

Example of using multiple background tasks:

def task_one():
    print("Executing task one")

def task_two():
    print("Executing task two")

@app.post("/multiple-tasks/")
async def multiple_tasks(background_tasks: BackgroundTasks):
    background_tasks.add_task(task_one)
    background_tasks.add_task(task_two)
    return {"message": "Multiple tasks are running in the background"}

In this example, both task_one and task_two are added as background tasks. They will be executed one after another in the background.

How do you send files or images in the background in FastAPI?

You can use background tasks to handle file uploads, image processing, or file downloads. For example, you can process an uploaded file or image in the background after sending a response to the user.

Example of sending files in the background:

from fastapi import FastAPI, BackgroundTasks, UploadFile

app = FastAPI()

def save_file(file: UploadFile):
    with open(f"uploaded_files/{file.filename}", "wb") as f:
        f.write(file.file.read())
    print(f"File {file.filename} saved")

@app.post("/upload/")
async def upload_file(file: UploadFile, background_tasks: BackgroundTasks):
    background_tasks.add_task(save_file, file)
    return {"message": "File will be processed in the background"}

In this example, the file is uploaded and processed in the background after the response is returned to the client.

How do you handle errors in background tasks?

Background tasks in FastAPI run asynchronously after the response is sent, so any errors that occur in the task do not affect the response to the client. However, you should handle errors within the task function itself using try-except blocks or logging to ensure you are aware of any issues that occur.

Example of handling errors in background tasks:

import logging

def process_data(data: dict):
    try:
        # Simulate processing data
        if "error" in data:
            raise ValueError("Error occurred while processing data")
        print(f"Processing data: {data}")
    except Exception as e:
        logging.error(f"Error in background task: {e}")

@app.post("/process/")
async def process_endpoint(data: dict, background_tasks: BackgroundTasks):
    background_tasks.add_task(process_data, data)
    return {"message": "Data processing started in the background"}

In this example, errors in the background task are handled with a try-except block, and any exceptions are logged.

How do you test background tasks in FastAPI?

When testing background tasks in FastAPI, you can mock the task function or inspect the output of the background task using testing tools like unittest or pytest. You can verify that the task was queued correctly and executed as expected.

Example of testing background tasks:

from fastapi.testclient import TestClient
from fastapi import BackgroundTasks

app = FastAPI()
client = TestClient(app)

def mock_task():
    print("Mock task executed")

@app.post("/test-task/")
async def test_task(background_tasks: BackgroundTasks):
    background_tasks.add_task(mock_task)
    return {"message": "Task added"}

def test_background_task():
    response = client.post("/test-task/")
    assert response.status_code == 200
    assert response.json() == {"message": "Task added"}

In this example, the test verifies that the background task was successfully added, and you can inspect the logs or use a mock function to check that the task executed correctly.

When should you use background tasks in FastAPI?

Background tasks are useful when you need to perform time-consuming operations that don't need to block the client's request. Examples include sending emails, processing files, updating logs, or executing non-urgent tasks that can be done after the response is sent to the user. They improve the responsiveness of your API by offloading heavy tasks to run in the background.

Some typical use cases for background tasks include:

  • Sending notification emails
  • Processing uploaded files
  • Logging or analytics updates
  • Performing database cleanups
Ads