✦

AI Reading

Quick summary of this article

This module covers advanced FastAPI features that help you build production-ready APIs. You will learn how to use dependency injection for cleaner code, middleware for request/response processing, background tasks for non-blocking operations, event handlers for startup/shutdown logic, CORS configuration for cross-origin requests, and custom headers for additional client-server communication.

• Dependency injection lets you share logic like database sessions and authentication across endpoints without repeating code, using Depends() with functions or classes.
• Middleware processes every request and response, useful for logging, adding headers, or timing requests with the @app.middleware("http") decorator.
• Background tasks run after sending the response, ideal for emails or file processing, using BackgroundTasks.add_task().
• Event handlers run code on startup or shutdown, such as initializing a database connection or cache, using @app.on_event("startup") and @app.on_event("shutdown").
• CORS configuration with CORSMiddleware allows frontend apps on different domains to access your API, and custom headers let you read or add extra information like API versioning.

Introduction

Welcome to Module 11 of the FastAPI Complete Course. You have already built APIs with path operations, request validation, and database integration. Now it is time to master the advanced features that make FastAPI a production-grade framework. In this chapter, we will explore FastAPI Advanced FastAPI Features that every professional developer must know: Dependency Injection, Middleware, Background Tasks, Event Handlers, CORS Configuration, and Custom Headers.

These features will help you write cleaner, more maintainable code, handle cross-origin requests, run tasks in the background, and control HTTP headers. By the end of this module, you will be able to build robust APIs that are ready for real-world deployment.

Dependency Injection

Dependency Injection (DI) is a design pattern where a component receives its dependencies from an external source rather than creating them internally. FastAPI has a powerful and intuitive DI system built-in. It allows you to share logic, database sessions, authentication, and configuration across your application without repeating code.

What is Dependency Injection in FastAPI?

In FastAPI, a dependency is a callable (function or class) that can be declared as a parameter in a path operation function. FastAPI automatically resolves and injects the required dependencies when the endpoint is called.

Common use cases include:

• Database session management
• Authentication and authorization
• Configuration settings
• Request validation and preprocessing

Creating a Simple Dependency Function

Let us start with a basic example. We will create a dependency that extracts a common query parameter limit from the request.

from fastapi import FastAPI, Depends, Query

app = FastAPI()

# Dependency function
def pagination_dependency(limit: int = Query(10, ge=1, le=100)):
    return limit

@app.get("/items/")
async def read_items(limit: int = Depends(pagination_dependency)):
    return {"limit": limit}

Explanation:

• pagination_dependency is a simple function that takes a query parameter limit with default value 10, and validates it to be between 1 and 100.
• In the path operation, we use Depends(pagination_dependency) to inject the result of the dependency into the limit parameter.
• FastAPI automatically calls the dependency, validates the input, and passes the result to the endpoint.

Class-Based Dependencies

For more complex scenarios, you can use classes as dependencies. Classes allow you to maintain state or encapsulate multiple related functions.

from fastapi import FastAPI, Depends
from typing import Optional

app = FastAPI()

class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/search/")
async def search_items(params: CommonQueryParams = Depends()):
    return {"q": params.q, "skip": params.skip, "limit": params.limit}

Explanation:

• CommonQueryParams is a class with three attributes: q, skip, and limit.
• When you use Depends() without arguments, FastAPI automatically detects the class and instantiates it by reading query parameters.
• This approach is cleaner when you have many related parameters.

Dependencies with Yield (Database Sessions)

FastAPI supports dependencies that use yield for setup and teardown logic. This is perfect for database sessions that need to be closed after the request.

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

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

@app.get("/users/")
async def read_users(db: Session = Depends(get_db)):
    # Use db session here
    return {"message": "Database session injected"}

Explanation:

• get_db creates a new database session, yields it for use in the endpoint, and then closes it in the finally block.
• FastAPI ensures the cleanup code runs even if an exception occurs.
• This pattern is the standard way to handle database sessions in FastAPI.

Middleware

Middleware is a layer that sits between the client request and your path operations. It can process every request before it reaches your endpoint and every response before it is sent back to the client. Middleware is useful for logging, adding headers, authentication checks, or request modification.

Creating Custom Middleware

FastAPI allows you to add middleware using the @app.middleware("http") decorator. The middleware function receives the request and a call_next function.

import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

@app.get("/")
async def root():
    return {"message": "Hello World"}

Explanation:

• The middleware records the start time before the request is processed.
• call_next(request) passes the request to the next middleware or the actual path operation.
• After the response is generated, we calculate the elapsed time and add it as a custom header X-Process-Time.
• This header will be present in every response from the API.

Practical Middleware Example: Request Logging

Let us build a middleware that logs every request method and URL.

import logging
from fastapi import FastAPI, Request

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    logger.info(f"Response status: {response.status_code}")
    return response

@app.get("/hello/")
async def hello():
    return {"message": "Logged request"}

Explanation:

• We set up basic logging to the console.
• The middleware logs the HTTP method and URL before processing.
• After the response is generated, it logs the status code.
• This is invaluable for debugging and monitoring in production.

Background Tasks

Sometimes you need to perform operations after sending the response to the client. For example, sending an email, processing a file, or updating a cache. FastAPI provides BackgroundTasks to handle such scenarios efficiently without blocking the response.

Using BackgroundTasks

FastAPI includes a BackgroundTasks class that you can inject into your path operations. You add tasks to it, and they run after the response is sent.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    with open("log.txt", mode="a") as log_file:
        log_file.write(f"{message}n")

@app.post("/send-notification/")
async def send_notification(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    return {"message": "Notification will be sent"}

Explanation:

• write_log is a regular function that writes a message to a file.
• BackgroundTasks is injected into the endpoint.
• background_tasks.add_task(write_log, ...) schedules the function to run after the response is returned.
• The client receives the response immediately, while the log writing happens in the background.

Practical Example: Email Simulation

Let us simulate sending a welcome email after user registration.

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

def send_welcome_email(email: str, username: str):
    # Simulate email sending (replace with actual email logic)
    print(f"Sending welcome email to {email} for user {username}")

class UserCreate(BaseModel):
    username: str
    email: str

@app.post("/register/")
async def register_user(user: UserCreate, background_tasks: BackgroundTasks):
    # Save user to database (omitted for brevity)
    background_tasks.add_task(send_welcome_email, user.email, user.username)
    return {"message": "User registered. Welcome email will be sent."}

Explanation:

• The endpoint accepts user data and immediately returns a response.
• The background task send_welcome_email runs asynchronously after the response.
• This pattern improves user experience by reducing response time.

Event Handlers

Event handlers allow you to run code when the application starts up or shuts down. This is useful for initializing connections (database, cache) or cleaning up resources.

Startup and Shutdown Events

FastAPI provides @app.on_event("startup") and @app.on_event("shutdown") decorators.

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    print("Application is starting up...")
    # Initialize database connection, load models, etc.

@app.on_event("shutdown")
async def shutdown_event():
    print("Application is shutting down...")
    # Close database connections, clean up resources

@app.get("/")
async def root():
    return {"message": "Event handlers active"}

Explanation:

• The startup event runs once when the application starts.
• The shutdown event runs when the application is gracefully terminated.
• You can use these to set up and tear down resources like database pools or external API clients.

Practical Example: Database Connection Pool

Let us create a simple in-memory cache that initializes on startup.

from fastapi import FastAPI

app = FastAPI()
cache = {}

@app.on_event("startup")
async def load_cache():
    global cache
    cache = {"app_name": "FastAPI Advanced Features", "version": "1.0"}
    print("Cache loaded on startup")

@app.get("/cache/")
async def get_cache():
    return cache

Explanation:

• The startup event initializes a global cache dictionary.
• This ensures that the cache is ready before any request is processed.
• In production, you might load data from a database or external API.

CORS Configuration

Cross-Origin Resource Sharing (CORS) is a security mechanism that allows or restricts resources on a web page to be requested from another domain. If your FastAPI backend serves a frontend on a different domain (e.g., React on localhost:3000, FastAPI on localhost:8000), you must configure CORS.

Setting Up CORS with CORSMiddleware

FastAPI provides CORSMiddleware from the starlette.middleware.cors module.

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
    "http://localhost:3000",    # React frontend
    "http://localhost:8080",    # Vue.js frontend
    "https://myfrontend.com",   # Production frontend
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def root():
    return {"message": "CORS is configured"}

Explanation:

• allow_origins specifies which domains are allowed to make requests.
• allow_credentials allows cookies and authentication headers.
• allow_methods=["*"] allows all HTTP methods (GET, POST, PUT, DELETE, etc.).
• allow_headers=["*"] allows all headers.
• In production, restrict origins to your actual frontend domain for security.

Common CORS Mistakes

• Using allow_origins=["*"] with allow_credentials=True — this is not allowed by browsers. You must specify explicit origins.
• Forgetting to add the middleware before other middleware or route definitions.
• Not including the correct port number in the origin.

Custom Headers

Custom headers allow you to pass additional information between the client and server. You can read custom headers from incoming requests and add custom headers to responses.

Reading Custom Headers from Requests

You can access custom headers using the Request object or by declaring them as parameters.

from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items/")
async def read_items(x_token: str = Header(None)):
    return {"X-Token": x_token}

Explanation:

• Header(None) tells FastAPI to extract the X-Token header from the request.
• The header name is converted to lowercase and hyphens become underscores by default.
• If the header is missing, the value will be None.

Adding Custom Headers to Responses

You can add custom headers to responses using the Response object or by returning a Response directly.

from fastapi import FastAPI, Response

app = FastAPI()

@app.get("/custom-header/")
async def custom_header():
    content = {"message": "Custom header example"}
    response = Response(content=content, media_type="application/json")
    response.headers["X-Custom-Header"] = "CustomValue123"
    return response

Explanation:

• We create a Response object with our JSON content.
• We then set a custom header X-Custom-Header on the response object.
• This header will be sent to the client.

Practical Example: API Versioning via Headers

Let us use a custom header to implement simple API versioning.

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()

@app.get("/version/")
async def get_version(api_version: str = Header("v1")):
    if api_version == "v1":
        return {"version": "1.0", "features": ["basic"]}
    elif api_version == "v2":
        return {"version": "2.0", "features": ["basic", "advanced"]}
    else:
        raise HTTPException(status_code=400, detail="Unsupported API version")

Explanation:

• The client sends an api-version header (FastAPI converts to api_version).
• Based on the header value, different responses are returned.
• This is a lightweight approach to versioning without changing URL paths.

Common Mistakes

Beginners often encounter these pitfalls when working with FastAPI Advanced FastAPI Features:

• Using Depends() incorrectly: Forgetting to import Depends or not using it as a default value. Always use param: Type = Depends(dependency).
• Blocking the event loop in background tasks: Background tasks run in the same event loop. If you have CPU-intensive tasks, use BackgroundTasks with async functions or offload to a task queue like Celery.
• Forgetting to close database sessions: In dependencies with yield, always close the session in the finally block to prevent connection leaks.
• Misconfiguring CORS: Using wildcard origins with credentials, or not including the scheme (http/https) in origins.
• Overusing middleware: Adding too many middleware layers can slow down your application. Keep middleware lightweight.

Practice Task

Now it is your turn to apply what you have learned. Build a small FastAPI application that includes the following:

1. A dependency class that extracts pagination parameters (page and per_page) with default values and validation.
2. A middleware that logs the request method, URL, and response status code to a file.
3. A background task that simulates sending a confirmation email after a POST request to /subscribe/.
4. A startup event that prints “Application started” and a shutdown event that prints “Application stopped”.
5. CORS configuration that allows requests from http://localhost:3000.
6. An endpoint that reads a custom header X-User-ID and returns it in the response.

Test your application using a tool like Postman or curl. Ensure all features work correctly.

Summary

In this module, you learned the essential FastAPI Advanced FastAPI Features that elevate your API development skills:

• Dependency Injection: Reusable logic for database sessions, authentication, and parameters using functions and classes.
• Middleware: Process requests and responses globally for logging, headers, or authentication.
• Background Tasks: Run operations after sending the response to improve user experience.
• Event Handlers: Initialize and clean up resources at application startup and shutdown.
• CORS Configuration: Enable cross-origin requests from frontend applications.
• Custom Headers: Read and write custom headers for versioning, tokens, or metadata.

These features are not just theoretical—they are used daily in production FastAPI applications. Master them, and you will be ready to build scalable, maintainable APIs.

FAQs

1. Can I use multiple dependencies in one endpoint?

Yes, you can declare multiple parameters with Depends(). FastAPI will resolve all dependencies and inject them in order. This is useful for combining authentication, database sessions, and query parameters.

2. Are background tasks suitable for long-running operations?

No, background tasks are designed for short operations (a few seconds). For long-running or CPU-intensive tasks, use a task queue like Celery or Redis Queue. Background tasks run in the same event loop and can block other requests.

3. How do I test middleware in FastAPI?

You can use FastAPI’s TestClient from starlette.testclient. The test client runs your middleware automatically. You can inspect response headers and status codes to verify middleware behavior.

4. What is the difference between middleware and dependencies?

Middleware processes every request and response globally, while dependencies are injected into specific endpoints. Middleware is best for cross-cutting concerns (logging, CORS), whereas dependencies handle per-endpoint logic (database access, authentication).

5. Can I add multiple CORS origins dynamically?

Yes, you can pass a list of origins to allow_origins. For dynamic origins, you can use a function that returns the allowed origins based on the request. However, for most applications, a static list is sufficient and more secure.

---

Next Up: In Module 12, we will dive into Testing and Debugging FastAPI Applications. You will learn how to write unit tests, use the TestClient, debug with logging and breakpoints, and ensure your API is robust and error-free. Keep building!

More Practical Examples

Let’s dive deeper into practical implementations of the advanced FastAPI features we’ve covered. These examples will help you see how these features work together in real-world scenarios.

Combining Dependency Injection with Custom Headers

Here’s a practical example that combines dependency injection with custom headers for API versioning and authentication:

from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optional

app = FastAPI()

# Dependency to extract and validate API version
def get_api_version(x_api_version: Optional[str] = Header(None)):
    if x_api_version is None:
        raise HTTPException(status_code=400, detail="X-API-Version header required")
    if x_api_version not in ["1.0", "2.0"]:
        raise HTTPException(status_code=400, detail="Unsupported API version")
    return x_api_version

# Dependency to validate API key
def validate_api_key(x_api_key: Optional[str] = Header(None)):
    if x_api_key is None:
        raise HTTPException(status_code=401, detail="API key required")
    # In production, check against a database
    valid_keys = {"key123", "key456", "key789"}
    if x_api_key not in valid_keys:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return x_api_key

@app.get("/users/")
async def get_users(
    api_version: str = Depends(get_api_version),
    api_key: str = Depends(validate_api_key)
):
    """Get users with version and key validation"""
    return {
        "api_version": api_version,
        "users": [
            {"id": 1, "name": "Alice"},
            {"id": 2, "name": "Bob"}
        ]
    }

This example shows how dependencies can validate headers before your endpoint logic runs. The get_api_version dependency ensures the client sends a valid version header, while validate_api_key checks authentication. This pattern keeps your endpoints clean and focused on business logic.

Background Tasks with Event Handlers

Here’s how to use background tasks alongside event handlers for database cleanup operations:

from fastapi import FastAPI, BackgroundTasks
from datetime import datetime
import asyncio

app = FastAPI()

# In-memory task log (use a database in production)
task_log = []

@app.on_event("startup")
async def startup_event():
    print(f"Server started at {datetime.now()}")
    # Initialize any resources
    task_log.clear()

@app.on_event("shutdown")
async def shutdown_event():
    print(f"Server shutting down at {datetime.now()}")
    # Clean up pending tasks
    pending = len([t for t in task_log if t["status"] == "pending"])
    if pending > 0:
        print(f"Warning: {pending} tasks still pending")

async def process_data(user_id: int, data: str):
    """Simulate long-running data processing"""
    await asyncio.sleep(5)  # Simulate processing time
    task_log.append({
        "user_id": user_id,
        "data": data,
        "status": "completed",
        "timestamp": datetime.now().isoformat()
    })
    print(f"Processed data for user {user_id}")

@app.post("/process/{user_id}")
async def create_process(
    user_id: int,
    data: str,
    background_tasks: BackgroundTasks
):
    """Start background data processing"""
    task_log.append({
        "user_id": user_id,
        "data": data,
        "status": "pending",
        "timestamp": datetime.now().isoformat()
    })
    background_tasks.add_task(process_data, user_id, data)
    return {"message": "Processing started", "user_id": user_id}

@app.get("/tasks/")
async def get_tasks():
    """View all tasks"""
    return {"tasks": task_log}

This example demonstrates how background tasks can handle time-consuming operations without blocking the response. The event handlers ensure proper initialization and cleanup of resources. Notice how the startup event clears the task log, and the shutdown event warns about pending tasks.

Middleware with CORS Configuration

Here’s a middleware that adds request timing and works alongside CORS:

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import time

app = FastAPI()

# Configure CORS first
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myfrontend.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["X-Process-Time"],
)

# Custom timing middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    
    # Log slow requests
    if process_time > 1.0:
        print(f"Slow request: {request.url.path} took {process_time:.2f}s")
    
    return response

@app.get("/slow-endpoint/")
async def slow_endpoint():
    """Simulate a slow endpoint"""
    time.sleep(2)
    return {"message": "This was slow"}

@app.get("/fast-endpoint/")
async def fast_endpoint():
    """Fast endpoint"""
    return {"message": "This was fast"}

This middleware adds a custom header showing how long each request took. It also logs slow requests for monitoring. The CORS configuration is set up first, then the custom middleware runs for every request. The expose_headers parameter in CORS is crucial—without it, browsers won’t allow JavaScript to read the X-Process-Time header.

Class-Based Example

While FastAPI primarily uses functions, you can organize advanced features using classes for better code organization and reusability. Here’s a complete example:

from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks, Request
from fastapi.middleware.cors import CORSMiddleware
from typing import Optional, List, Dict
from datetime import datetime
import asyncio
import json

class APIConfig:
    """Configuration class for API settings"""
    def __init__(self, version: str = "1.0", debug: bool = False):
        self.version = version
        self.debug = debug
        self.allowed_origins = ["http://localhost:3000"]
        self.api_keys = {"admin123": "admin", "user456": "user"}
    
    def validate_key(self, api_key: str) -> str:
        """Validate API key and return role"""
        if api_key not in self.api_keys:
            raise HTTPException(status_code=403, detail="Invalid API key")
        return self.api_keys[api_key]

class DataProcessor:
    """Class for processing data with background tasks"""
    def __init__(self, config: APIConfig):
        self.config = config
        self.processing_log: List[Dict] = []
    
    async def process_item(self, item_id: int, data: Dict):
        """Simulate data processing"""
        await asyncio.sleep(3)
        result = {
            "item_id": item_id,
            "original_data": data,
            "processed_at": datetime.now().isoformat(),
            "api_version": self.config.version
        }
        self.processing_log.append(result)
        return result
    
    def get_log(self) -> List[Dict]:
        return self.processing_log

# Initialize application
config = APIConfig(version="2.0", debug=True)
processor = DataProcessor(config)

app = FastAPI()

# Configure CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=config.allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Dependency using class method
def get_api_key(request: Request) -> str:
    api_key = request.headers.get("X-API-Key")
    if not api_key:
        raise HTTPException(status_code=401, detail="API key required")
    return config.validate_key(api_key)

@app.post("/process/{item_id}")
async def start_processing(
    item_id: int,
    data: Dict,
    background_tasks: BackgroundTasks,
    role: str = Depends(get_api_key)
):
    """Start background processing (admin only)"""
    if role != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    
    background_tasks.add_task(processor.process_item, item_id, data)
    return {
        "message": "Processing started",
        "item_id": item_id,
        "role": role
    }

@app.get("/logs/")
async def get_processing_logs(
    role: str = Depends(get_api_key)
):
    """Get processing logs"""
    return {
        "logs": processor.get_log(),
        "count": len(processor.get_log())
    }

@app.get("/config/")
async def get_config():
    """Get API configuration (no auth needed)"""
    return {
        "version": config.version,
        "debug": config.debug,
        "allowed_origins": config.allowed_origins
    }

This class-based approach offers several advantages:

• Configuration Management: The APIConfig class centralizes all settings, making it easy to modify without changing endpoint code.
• State Management: The DataProcessor class maintains state across requests, useful for caching or logging.
• Dependency Injection: The get_api_key function uses the config class to validate keys, demonstrating how classes integrate with FastAPI’s dependency system.
• Role-Based Access: The example shows how to implement different access levels using the validated role.

Step-by-Step Exercise

Let’s build a complete feature that uses all the advanced concepts. Follow these steps to create an API that processes user uploads with validation, background processing, and logging.

Step 1: Project Setup

Create a new directory and install FastAPI:

mkdir advanced-api
cd advanced-api
python -m venv venv
source venv/bin/activate  # On Windows: venvScriptsactivate
pip install fastapi uvicorn python-multipart

Step 2: Create the Main Application

Create main.py with the following code:

from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks, File, UploadFile, Header
from fastapi.middleware.cors import CORSMiddleware
from typing import Optional, List
from datetime import datetime
import json
import os

app = FastAPI(title="Advanced File Processor")

# CORS configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# In-memory storage for uploaded files metadata
uploaded_files = []

# Dependency to validate user token
async def validate_user(authorization: Optional[str] = Header(None)):
    if not authorization:
        raise HTTPException(status_code=401, detail="Authorization header required")
    # Simple token validation (use JWT in production)
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid authorization format")
    token = authorization.replace("Bearer ", "")
    if token != "valid-token-123":
        raise HTTPException(status_code=403, detail="Invalid token")
    return {"user": "test_user", "token": token}

# Background task to process uploaded file
async def process_file(file_id: int, filename: str, content: bytes):
    """Simulate file processing"""
    await asyncio.sleep(2)  # Simulate processing time
    # Update file status
    for file_info in uploaded_files:
        if file_info["id"] == file_id:
            file_info["status"] = "processed"
            file_info["processed_at"] = datetime.now().isoformat()
            file_info["size"] = len(content)
            break

@app.post("/upload/")
async def upload_file(
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...),
    user: dict = Depends(validate_user)
):
    """Upload a file for processing"""
    # Read file content
    content = await file.read()
    
    # Create file metadata
    file_id = len(uploaded_files) + 1
    file_info = {
        "id": file_id,
        "filename": file.filename,
        "content_type": file.content_type,
        "uploaded_at": datetime.now().isoformat(),
        "status": "pending",
        "user": user["user"]
    }
    uploaded_files.append(file_info)
    
    # Start background processing
    background_tasks.add_task(process_file, file_id, file.filename, content)
    
    return {
        "message": "File uploaded successfully",
        "file_id": file_id,
        "filename": file.filename,
        "status": "pending"
    }

@app.get("/files/")
async def list_files(user: dict = Depends(validate_user)):
    """List all uploaded files"""
    return {
        "files": uploaded_files,
        "total": len(uploaded_files)
    }

@app.get("/files/{file_id}")
async def get_file_status(
    file_id: int,
    user: dict = Depends(validate_user)
):
    """Get status of a specific file"""
    for file_info in uploaded_files:
        if file_info["id"] == file_id:
            return file_info
    raise HTTPException(status_code=404, detail="File not found")

Step 3: Add Event Handlers

Add these event handlers to your main.py:

@app.on_event("startup")
async def startup():
    """Initialize on startup"""
    print(f"Server started at {datetime.now()}")
    print("File processor API ready")
    # Clear any previous data
    uploaded_files.clear()

@app.on_event("shutdown")
async def shutdown():
    """Cleanup on shutdown"""
    print(f"Server shutting down at {datetime.now()}")
    pending_files = [f for f in uploaded_files if f["status"] == "pending"]
    if pending_files:
        print(f"Warning: {len(pending_files)} files still pending")

Step 4: Run and Test

Start the server:

uvicorn main:app --reload

Test the API using curl commands:

# Upload a file
curl -X POST http://localhost:8000/upload/ 
  -H "Authorization: Bearer valid-token-123" 
  -F "file=@test.txt"

# Check file status
curl http://localhost:8000/files/1 
  -H "Authorization: Bearer valid-token-123"

# List all files
curl http://localhost:8000/files/ 
  -H "Authorization: Bearer valid-token-123"

Step 5: Add Middleware for Logging

Add this middleware to log all requests:

@app.middleware("http")
async def log_requests(request, call_next):
    start_time = datetime.now()
    response = await call_next(request)
    process_time = (datetime.now() - start_time).total_seconds()
    print(f"{request.method} {request.url.path} - {response.status_code} - {process_time:.3f}s")
    return response

This exercise demonstrates how all the advanced features work together: dependency injection for authentication, background tasks for file processing, event handlers for lifecycle management, CORS for frontend access, custom headers for authorization, and middleware for logging.

Interview and Job Use Cases

Understanding these advanced FastAPI features is crucial for technical interviews and real-world development. Here’s how they’re typically used:

Common Interview Questions

1. “How do you handle authentication in FastAPI?”
   Use dependency injection with custom headers. Create a dependency function that extracts and validates tokens from the Authorization header, then inject it into protected endpoints.
2. “How would you implement rate limiting?”
   Create middleware that tracks request counts using in-memory storage or Redis. The middleware checks the count before processing the request and returns a 429 status code if exceeded.
3. “How do you handle long-running tasks without blocking?”
   Use BackgroundTasks for simple operations or a task queue like Celery for complex workflows. The endpoint returns immediately while the task runs in the background.

Real-World Job Scenarios

E-commerce API: Use dependency injection to validate user sessions, middleware to track API usage for billing, background tasks to process order confirmations and send emails, and event handlers to initialize payment gateways on startup.

Content Management System: Implement custom headers for API versioning, CORS to allow multiple frontend domains, background tasks for image processing and thumbnail generation, and dependency injection for role-based access control.

Data Analytics Platform: Use middleware to log all data queries for auditing, background tasks to run complex calculations, event handlers to connect to databases on startup, and dependency injection to validate API keys for different subscription tiers.

Performance Considerations

When using these features in production:

• Dependency Injection: Cache expensive operations (like database lookups) within dependencies
• Middleware: Keep middleware lightweight—avoid heavy computation or I/O operations
• Background Tasks: Use a proper task queue for critical operations that must survive server restarts
• Event Handlers: Ensure startup handlers complete quickly to avoid delaying the first request

Extra Beginner FAQs

Q: What’s the difference between middleware and dependencies?

Middleware runs for every request to your application, regardless of the endpoint. Dependencies only run for specific endpoints where they’re injected. Use middleware for global concerns like logging or CORS, and dependencies for endpoint-specific needs like authentication.

Q: Can I use multiple background tasks in one endpoint?

Yes! You can add multiple tasks to the same BackgroundTasks object. They’ll run in the order you add them, but they don’t wait for each other to complete. If you need sequential execution, use a single task that calls multiple functions.

Q: Why do I need to configure CORS if my frontend and backend are on the same server?

Even on the same server, they might be on different ports (e.g., frontend on port 3000, backend on port 8000). Browsers consider different ports as different origins, so CORS is still required. For production, use the same domain and port, or configure CORS properly.

Q: How do I test background tasks during development?

You can reduce the sleep time in your tasks for faster testing. Alternatively, create a synchronous version of your task for testing and swap it out in production. FastAPI’s BackgroundTasks also works with synchronous functions if you don’t need async.

Q: What happens to background tasks if the server crashes?

Background tasks are lost if the server crashes because they run in the same process. For critical tasks, use a message queue like RabbitMQ or Redis with a task queue like Celery. These systems can retry failed tasks and survive server restarts.

Q: Can I pass data from middleware to my endpoint?

Yes! You can attach data to the request.state object in middleware, then access it in your endpoint. For example: request.state.user = {"id": 1} in middleware, then request.state.user in the endpoint. Dependencies are usually a cleaner approach for this.

Q: How do I handle file uploads with background tasks?

Read the file content in the endpoint and pass it to the background task. Don’t try to access the UploadFile object in the background task because the request has already ended. Store the file to disk or database first, then process the stored file in the background.

These advanced features make FastAPI incredibly powerful for building production-ready APIs. Practice combining them in different ways to solve real-world problems, and you’ll be well-prepared for both interviews and job tasks.