Skip to main content

FastAPI Dependency Yielding

Introduction​

In FastAPI, dependencies are a powerful way to share code logic across multiple request handlers. While basic dependencies return values, FastAPI also offers a more advanced pattern called dependency yielding. This feature allows you to execute code both before a request is processed and after the response has been delivered.

Dependency yielding uses Python's yield statement and behaves similar to context managers, making it perfect for scenarios where you need to:

  • Establish and close database connections
  • Acquire and release resources
  • Set up and tear down test environments
  • Perform cleanup operations
  • Track execution time or log request flows

This guide will explore how dependency yielding works, its benefits, and practical examples to incorporate it in your FastAPI applications.

How Dependency Yielding Works​

Basic Concept​

Unlike regular dependencies that use return, yielding dependencies use yield to:

  1. Execute code before handling the request
  2. Yield the value to be injected
  3. Resume execution after the response has been delivered
  4. Perform cleanup operations

Here's a simple comparison:

python
# Regular dependency
def regular_dependency():
    value = "I'm a regular dependency"
    return value  # Returns and ends here

# Yielding dependency
def yielding_dependency():
    value = "I'm a yielding dependency"
    print("Before yield - setting up")
    yield value  # Pauses here, returns value, and continues after request is complete
    print("After yield - cleaning up")  # Runs after response is sent

The Request Lifecycle with Yielding Dependencies​

When you use a yielding dependency, the request lifecycle follows these steps:

  1. Request arrives at your FastAPI application
  2. The dependency is executed until it reaches a yield statement
  3. The yielded value is passed to the path operation function
  4. The path operation function is executed and a response is generated
  5. The response is sent to the client
  6. Execution of the dependency resumes from the yield statement
  7. Any cleanup code after the yield statement is executed

Basic Examples of Dependency Yielding​

Example 1: Simple Yielding Dependency​

python
from fastapi import Depends, FastAPI

app = FastAPI()

async def get_db():
    print("Connecting to database...")
    db = {"data": "some database connection"}  # Simulate DB connection
    yield db  # Provide the DB connection
    print("Closing database connection...")  # This runs after the response is sent

@app.get("/items/")
async def read_items(db: dict = Depends(get_db)):
    return {"database": db, "items": ["Item1", "Item2"]}

When you make a request to /items/, the server logs will show:

Connecting to database...
Closing database connection...

And the response will be:

json
{
  "database": {"data": "some database connection"},
  "items": ["Item1", "Item2"]
}

Example 2: Error Handling with try/finally​

One of the key benefits of yielding dependencies is that cleanup code runs even if there's an exception during request processing:

python
from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()

async def get_db():
    print("Connecting to database...")
    db = {"data": "some database connection"}
    try:
        yield db
    finally:
        print("Closing database connection (even if there was an error)...")

@app.get("/items/{item_id}")
async def read_item(item_id: int, db: dict = Depends(get_db)):
    if item_id == 0:
        raise HTTPException(status_code=400, detail="Item ID cannot be zero")
    return {"item_id": item_id, "database": db}

Even when an exception occurs (e.g., requesting /items/0), the database connection will still be properly closed.

Practical Applications​

Database Session Management​

One of the most common uses of yielding dependencies is database session management:

python
from fastapi import Depends, FastAPI
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

app = FastAPI()

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

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
        # Changes are committed if no exceptions occurred
        db.commit()
    except:
        # Roll back changes if an exception occurred
        db.rollback()
        raise
    finally:
        # Always close the session
        db.close()

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Performance Monitoring​

Yielding dependencies can be used to track the execution time of requests:

python
import time
from fastapi import Depends, FastAPI, Request

app = FastAPI()

async def timer():
    start = time.time()
    # Store start time in a variable that can be accessed later
    request_timer = {"start": start}
    yield request_timer
    # Calculate execution time after response is sent
    execution_time = time.time() - start
    print(f"Request took {execution_time:.4f} seconds to process")

@app.get("/slow-operation/")
async def slow_operation(timer_data: dict = Depends(timer)):
    # Simulate a time-consuming operation
    time.sleep(1)
    return {"message": "Operation completed"}

Resource Acquisition and Release​

For scenarios where you need to acquire and release resources:

python
from fastapi import Depends, FastAPI
import aiofiles

app = FastAPI()

async def get_file_handler():
    # Open a file for reading
    print("Opening file...")
    file = await aiofiles.open("data.txt", mode="r")
    try:
        yield file
    finally:
        # Close the file even if there's an error
        print("Closing file...")
        await file.close()

@app.get("/file-content/")
async def read_file_content(file = Depends(get_file_handler)):
    content = await file.read()
    return {"content": content}

Authentication with Cleanup​

You can implement a token verification system with cleanup tasks:

python
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_token(token: str = Depends(oauth2_scheme)):
    print(f"Verifying token: {token}")
    # Simulate token verification
    if token != "valid_token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )
    
    # Record authentication in an audit log
    yield {"user_id": "user123"}
    
    # After response, log the completed request
    print(f"Request with token {token} completed")

@app.get("/protected-resource/")
async def get_protected_resource(user_data: dict = Depends(verify_token)):
    return {"message": "This is a protected resource", "user_id": user_data["user_id"]}

Advanced Patterns​

Nested Dependencies with Yielding​

You can combine yielding dependencies with other dependencies:

python
from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters():
    return {"q": "default", "skip": 0, "limit": 100}

async def pagination(commons: dict = Depends(common_parameters)):
    pagination_params = {
        "skip": commons["skip"],
        "limit": commons["limit"]
    }
    print("Setting up pagination...")
    yield pagination_params
    print("Pagination context finished")

@app.get("/items/")
async def read_items(pagination_params: dict = Depends(pagination)):
    return {
        "pagination": pagination_params,
        "items": ["Item1", "Item2", "Item3"]
    }

Using Yielding Dependencies with WebSockets​

Yielding dependencies are also useful for WebSocket connections:

python
from fastapi import Depends, FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

async def websocket_auth():
    print("Authenticating WebSocket connection...")
    user_id = "user123"  # In a real app, this would verify credentials
    try:
        yield {"user_id": user_id}
        print(f"WebSocket connection for user {user_id} closed normally")
    except WebSocketDisconnect:
        print(f"WebSocket for user {user_id} disconnected unexpectedly")

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, auth: dict = Depends(websocket_auth)):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message received from user {auth['user_id']}: {data}")
    except WebSocketDisconnect:
        # Will be caught by the dependency
        raise

Best Practices and Tips​

1. Always Use try/finally for Critical Cleanup​

For important resources that must be released, always use try/finally:

python
async def get_resource():
    resource = acquire_expensive_resource()
    try:
        yield resource
    finally:
        # This will always execute, even if exceptions occur
        release_resource(resource)

2. Keep Dependencies Focused​

Each dependency should have a single responsibility:

python
# Good: Single responsibility
async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Avoid: Too many responsibilities
async def get_db_and_validate_user_and_log():
    # This does too many things
    db = SessionLocal()
    # Validate user...
    # Log request...
    yield db
    # Close db...
    # Log response...

3. Avoid Long-Running Tasks After Yield​

After the yield statement, avoid performing long operations that would delay the next request:

python
# Bad practice
async def log_request():
    yield None
    # This delays the next request
    await asyncio.sleep(5)  # Don't do this!
    
# Better approach
async def log_request():
    yield None
    # Use background tasks for anything time-consuming
    background_tasks.add_task(process_logs)

4. Leveraging FastAPI's BackgroundTasks​

For operations after the response that might take time:

python
from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()

async def get_notification_manager(background_tasks: BackgroundTasks):
    yield {"status": "ready"}
    # Schedule work to happen in the background after the response
    background_tasks.add_task(send_notifications)

def send_notifications():
    # This runs in the background after the response is sent
    print("Sending notifications...")

@app.get("/trigger-notification/")
async def trigger(manager: dict = Depends(get_notification_manager)):
    return {"message": "Notification triggered"}

Common Issues and How to Solve Them​

1. Dependencies Not Cleaning Up​

Issue: Cleanup code doesn't seem to run.

Solution: Make sure you're using yield exactly once. Multiple yields or returns will break the pattern.

python
# Problematic (multiple yields)
async def broken_dependency():
    yield {"message": "first"}  # Only code up to here runs in cleanup
    yield {"message": "second"}  # Never reached in cleanup
    print("Cleanup")  # Never executed

# Fixed version
async def fixed_dependency():
    yield {"message": "only one yield"}
    print("This cleanup will run")

2. Exceptions in Yielding Dependencies​

Issue: Exceptions in post-yield code aren't visible to clients.

Solution: Log these exceptions, as they won't be returned to the client:

python
import logging

logger = logging.getLogger(__name__)

async def safe_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        try:
            db.close()
        except Exception as e:
            # This exception won't reach the client, so log it
            logger.error(f"Error closing database connection: {e}")

Summary​

FastAPI's dependency yielding is a powerful feature that lets you execute code both before and after request handling. This pattern is particularly useful for:

  • Resource management (opening/closing database connections)
  • Cleanup operations
  • Performance monitoring
  • Authentication with logging
  • Error handling with guaranteed cleanup

By using the yield keyword in your dependency functions, you can create more maintainable, efficient, and robust FastAPI applications.

Additional Resources​

Exercises​

  1. Create a yielding dependency that logs the start time and end time of each request.
  2. Implement a database dependency that rolls back transactions if any exceptions occur.
  3. Build a caching system using yielding dependencies where cache entries are cleaned up after the request.
  4. Create a rate limiter dependency that tracks and limits requests per user.
  5. Implement a dependency that acquires a connection from a connection pool and returns it after the request.

By mastering dependency yielding, you'll write more maintainable FastAPI applications with proper resource management and cleaner code organization.


If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)