FastAPI Dependency Injection

FastAPI依赖注入

Master FastAPI's dependency injection system for building modular, testable APIs with reusable dependencies.

掌握FastAPI的依赖注入系统，构建具备可复用依赖项的模块化、可测试API。

Basic Dependencies

基础依赖项

Simple dependency injection patterns in FastAPI.

python

from fastapi import Depends, FastAPI

app = FastAPI()

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

@app.get('/users')
async def get_users(db = Depends(get_db)):
    return await db.query('SELECT * FROM users')

FastAPI中的简单依赖注入模式。

python

from fastapi import Depends, FastAPI

app = FastAPI()

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

@app.get('/users')
async def get_users(db = Depends(get_db)):
    return await db.query('SELECT * FROM users')

Simple function dependency

def common_parameters(q: str = None, skip: int = 0, limit: int = 100): return {'q': q, 'skip': skip, 'limit': limit}

@app.get('/items') async def read_items(commons: dict = Depends(common_parameters)): return commons

def common_parameters(q: str = None, skip: int = 0, limit: int = 100): return {'q': q, 'skip': skip, 'limit': limit}

@app.get('/items') async def read_items(commons: dict = Depends(common_parameters)): return commons

Async dependencies

async def get_async_db(): db = await AsyncDatabase.connect() try: yield db finally: await db.disconnect()

@app.get('/async-users') async def get_async_users(db = Depends(get_async_db)): return await db.fetch_all('SELECT * FROM users')

undefined

async def get_async_db(): db = await AsyncDatabase.connect() try: yield db finally: await db.disconnect()

@app.get('/async-users') async def get_async_users(db = Depends(get_async_db)): return await db.fetch_all('SELECT * FROM users')

undefined

Dependency Scopes

依赖项作用域

Understand dependency lifecycle and caching.

python

from fastapi import Depends

了解依赖项的生命周期与缓存机制。

python

from fastapi import Depends

Request-scoped dependency (default, cached per request)

def get_current_user(token: str = Depends(oauth2_scheme)): user = decode_token(token) return user

Multiple uses in same request reuse the same instance

@app.get('/profile') async def get_profile( user1 = Depends(get_current_user), user2 = Depends(get_current_user) # Same instance as user1 ): assert user1 is user2 # True return user1

No caching (use_cache=False)

def get_fresh_data(use_cache: bool = False): return fetch_from_api()

@app.get('/data') async def get_data(data = Depends(get_fresh_data, use_cache=False)): return data # Fetches fresh data each time

def get_fresh_data(use_cache: bool = False): return fetch_from_api()

@app.get('/data') async def get_data(data = Depends(get_fresh_data, use_cache=False)): return data # Fetches fresh data each time

Singleton pattern (application scope)

class Settings: def init(self): self.app_name = "MyApp" self.admin_email = "admin@example.com"

settings = Settings() # Created once at startup

def get_settings(): return settings

@app.get('/settings') async def read_settings(settings: Settings = Depends(get_settings)): return settings

undefined

class Settings: def init(self): self.app_name = "MyApp" self.admin_email = "admin@example.com"

settings = Settings() # Created once at startup

def get_settings(): return settings

@app.get('/settings') async def read_settings(settings: Settings = Depends(get_settings)): return settings

undefined

Sub-Dependencies

子依赖项

Build complex dependency chains.

python

from fastapi import Depends, HTTPException, status

构建复杂的依赖项链。

python

from fastapi import Depends, HTTPException, status

Sub-dependency chain

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

def get_current_user( token: str = Depends(oauth2_scheme), db = Depends(get_db) ): user = db.query_one('SELECT * FROM users WHERE token = ?', token) if not user: raise HTTPException(status_code=401, detail='Invalid token') return user

def get_current_active_user( current_user = Depends(get_current_user) ): if not current_user.is_active: raise HTTPException(status_code=400, detail='Inactive user') return current_user

def get_admin_user( current_user = Depends(get_current_active_user) ): if not current_user.is_admin: raise HTTPException(status_code=403, detail='Not authorized') return current_user

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

def get_current_user( token: str = Depends(oauth2_scheme), db = Depends(get_db) ): user = db.query_one('SELECT * FROM users WHERE token = ?', token) if not user: raise HTTPException(status_code=401, detail='Invalid token') return user

def get_current_active_user( current_user = Depends(get_current_user) ): if not current_user.is_active: raise HTTPException(status_code=400, detail='Inactive user') return current_user

def get_admin_user( current_user = Depends(get_current_active_user) ): if not current_user.is_admin: raise HTTPException(status_code=403, detail='Not authorized') return current_user

Use in endpoint

@app.delete('/users/{user_id}') async def delete_user( user_id: int, admin = Depends(get_admin_user), db = Depends(get_db) ): await db.execute('DELETE FROM users WHERE id = ?', user_id) return {'status': 'deleted'}

undefined

@app.delete('/users/{user_id}') async def delete_user( user_id: int, admin = Depends(get_admin_user), db = Depends(get_db) ): await db.execute('DELETE FROM users WHERE id = ?', user_id) return {'status': 'deleted'}

undefined

Class-Based Dependencies

基于类的依赖项

Use classes for stateful dependencies.

python

from fastapi import Depends

class Database:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.connection = None

    async def connect(self):
        self.connection = await create_connection(self.connection_string)
        return self

    async def disconnect(self):
        if self.connection:
            await self.connection.close()

    async def fetch_all(self, query: str):
        return await self.connection.fetch_all(query)

使用类实现有状态的依赖项。

python

from fastapi import Depends

class Database:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.connection = None

    async def connect(self):
        self.connection = await create_connection(self.connection_string)
        return self

    async def disconnect(self):
        if self.connection:
            await self.connection.close()

    async def fetch_all(self, query: str):
        return await self.connection.fetch_all(query)

Callable class (using call)

class DatabaseDependency: def init(self): self.db = None

async def __call__(self):
    if not self.db:
        self.db = Database('postgresql://localhost/db')
        await self.db.connect()
    return self.db

db_dependency = DatabaseDependency()

@app.get('/users') async def get_users(db = Depends(db_dependency)): return await db.fetch_all('SELECT * FROM users')

class DatabaseDependency: def init(self): self.db = None

async def __call__(self):
    if not self.db:
        self.db = Database('postgresql://localhost/db')
        await self.db.connect()
    return self.db

db_dependency = DatabaseDependency()

@app.get('/users') async def get_users(db = Depends(db_dependency)): return await db.fetch_all('SELECT * FROM users')

Class with initialization parameters

class Pagination: def init(self, skip: int = 0, limit: int = 100): self.skip = skip self.limit = limit

@app.get('/items') async def get_items(pagination: Pagination = Depends()): return {'skip': pagination.skip, 'limit': pagination.limit}

class Pagination: def init(self, skip: int = 0, limit: int = 100): self.skip = skip self.limit = limit

@app.get('/items') async def get_items(pagination: Pagination = Depends()): return {'skip': pagination.skip, 'limit': pagination.limit}

Advanced class-based dependency with configuration

class ServiceClient: def init( self, api_key: str, timeout: int = 30, max_retries: int = 3 ): self.api_key = api_key self.timeout = timeout self.max_retries = max_retries self.client = None

async def __call__(self):
    if not self.client:
        self.client = await create_http_client(
            api_key=self.api_key,
            timeout=self.timeout
        )
    return self.client

class ServiceClient: def init( self, api_key: str, timeout: int = 30, max_retries: int = 3 ): self.api_key = api_key self.timeout = timeout self.max_retries = max_retries self.client = None

async def __call__(self):
    if not self.client:
        self.client = await create_http_client(
            api_key=self.api_key,
            timeout=self.timeout
        )
    return self.client

Factory function for configurable class-based dependency

def create_service_dependency(api_key: str): return ServiceClient(api_key=api_key, timeout=60)

service = create_service_dependency('my-api-key')

@app.get('/external-data') async def get_external_data(client = Depends(service)): return await client.fetch('/data')

undefined

def create_service_dependency(api_key: str): return ServiceClient(api_key=api_key, timeout=60)

service = create_service_dependency('my-api-key')

@app.get('/external-data') async def get_external_data(client = Depends(service)): return await client.fetch('/data')

undefined

Generator Dependencies for Cleanup

用于资源清理的生成器依赖项

Use generator functions to ensure proper resource cleanup.

python

from contextlib import asynccontextmanager
from fastapi import Depends
import httpx

使用生成器函数确保资源被正确清理。

python

from contextlib import asynccontextmanager
from fastapi import Depends
import httpx

Database connection with cleanup

async def get_db_connection(): connection = await Database.connect('postgresql://localhost/db') try: yield connection finally: await connection.close() print('Database connection closed')

HTTP client with cleanup

async def get_http_client(): async with httpx.AsyncClient(timeout=30.0) as client: yield client # Client automatically closed after yield

@app.get('/external-api') async def call_external_api(client = Depends(get_http_client)): response = await client.get('https://api.example.com/data') return response.json()

async def get_http_client(): async with httpx.AsyncClient(timeout=30.0) as client: yield client # Client automatically closed after yield

@app.get('/external-api') async def call_external_api(client = Depends(get_http_client)): response = await client.get('https://api.example.com/data') return response.json()

File handle with cleanup

async def get_file_writer(filename: str): file = await aiofiles.open(filename, mode='a') try: yield file finally: await file.close()

Multiple resources with cleanup

async def get_resources(): db = await Database.connect('postgresql://localhost/db') cache = await Redis.connect('redis://localhost') logger = Logger('app')

try:
    yield {'db': db, 'cache': cache, 'logger': logger}
finally:
    await cache.close()
    await db.close()
    logger.shutdown()

@app.post('/process') async def process_data( data: dict, resources = Depends(get_resources) ): db = resources['db'] cache = resources['cache'] logger = resources['logger']

logger.info('Processing data')
result = await db.save(data)
await cache.invalidate('data_cache')
return result

async def get_resources(): db = await Database.connect('postgresql://localhost/db') cache = await Redis.connect('redis://localhost') logger = Logger('app')

try:
    yield {'db': db, 'cache': cache, 'logger': logger}
finally:
    await cache.close()
    await db.close()
    logger.shutdown()

@app.post('/process') async def process_data( data: dict, resources = Depends(get_resources) ): db = resources['db'] cache = resources['cache'] logger = resources['logger']

logger.info('Processing data')
result = await db.save(data)
await cache.invalidate('data_cache')
return result

Context manager as dependency

@asynccontextmanager async def transaction_context(db = Depends(get_db)): async with db.begin() as transaction: try: yield transaction await transaction.commit() except Exception: await transaction.rollback() raise

async def get_transaction(db = Depends(get_db)): async with transaction_context(db) as txn: yield txn

@app.post('/transfer') async def transfer_funds( from_account: int, to_account: int, amount: float, txn = Depends(get_transaction) ): await txn.execute( 'UPDATE accounts SET balance = balance - ? WHERE id = ?', amount, from_account ) await txn.execute( 'UPDATE accounts SET balance = balance + ? WHERE id = ?', amount, to_account ) return {'status': 'success'}

undefined

@asynccontextmanager async def transaction_context(db = Depends(get_db)): async with db.begin() as transaction: try: yield transaction await transaction.commit() except Exception: await transaction.rollback() raise

async def get_transaction(db = Depends(get_db)): async with transaction_context(db) as txn: yield txn

@app.post('/transfer') async def transfer_funds( from_account: int, to_account: int, amount: float, txn = Depends(get_transaction) ): await txn.execute( 'UPDATE accounts SET balance = balance - ? WHERE id = ?', amount, from_account ) await txn.execute( 'UPDATE accounts SET balance = balance + ? WHERE id = ?', amount, to_account ) return {'status': 'success'}

undefined

Dependency Chains and Sub-Dependencies

依赖项链与子依赖项

Build complex dependency hierarchies.

python

from fastapi import Depends, HTTPException
from typing import Optional

构建复杂的依赖项层级结构。

python

from fastapi import Depends, HTTPException
from typing import Optional

Level 1: Configuration

def get_config(): return { 'database_url': 'postgresql://localhost/db', 'redis_url': 'redis://localhost', 'secret_key': 'super-secret' }

Level 2: Infrastructure (depends on config)

def get_db(config: dict = Depends(get_config)): db = Database(config['database_url']) try: yield db finally: db.close()

def get_cache(config: dict = Depends(get_config)): cache = Redis(config['redis_url']) try: yield cache finally: cache.close()

def get_db(config: dict = Depends(get_config)): db = Database(config['database_url']) try: yield db finally: db.close()

def get_cache(config: dict = Depends(get_config)): cache = Redis(config['redis_url']) try: yield cache finally: cache.close()

Level 3: Authentication (depends on infrastructure)

def get_token_from_header(authorization: str = Header(None)): if not authorization: raise HTTPException(status_code=401, detail='Missing token') scheme, token = authorization.split() if scheme.lower() != 'bearer': raise HTTPException(status_code=401, detail='Invalid scheme') return token

def verify_token( token: str = Depends(get_token_from_header), config: dict = Depends(get_config) ): try: payload = jwt.decode( token, config['secret_key'], algorithms=['HS256'] ) return payload except JWTError: raise HTTPException(status_code=401, detail='Invalid token')

def get_current_user( payload: dict = Depends(verify_token), db = Depends(get_db) ): user_id = payload.get('user_id') user = db.query_one('SELECT * FROM users WHERE id = ?', user_id) if not user: raise HTTPException(status_code=404, detail='User not found') return user

def get_token_from_header(authorization: str = Header(None)): if not authorization: raise HTTPException(status_code=401, detail='Missing token') scheme, token = authorization.split() if scheme.lower() != 'bearer': raise HTTPException(status_code=401, detail='Invalid scheme') return token

def verify_token( token: str = Depends(get_token_from_header), config: dict = Depends(get_config) ): try: payload = jwt.decode( token, config['secret_key'], algorithms=['HS256'] ) return payload except JWTError: raise HTTPException(status_code=401, detail='Invalid token')

def get_current_user( payload: dict = Depends(verify_token), db = Depends(get_db) ): user_id = payload.get('user_id') user = db.query_one('SELECT * FROM users WHERE id = ?', user_id) if not user: raise HTTPException(status_code=404, detail='User not found') return user

Level 4: Authorization (depends on authentication)

def get_active_user(user = Depends(get_current_user)): if not user.is_active: raise HTTPException(status_code=403, detail='Inactive user') return user

def require_permission(permission: str): def permission_checker(user = Depends(get_active_user)): if permission not in user.permissions: raise HTTPException( status_code=403, detail=f'Permission {permission} required' ) return user return permission_checker

def get_active_user(user = Depends(get_current_user)): if not user.is_active: raise HTTPException(status_code=403, detail='Inactive user') return user

def require_permission(permission: str): def permission_checker(user = Depends(get_active_user)): if permission not in user.permissions: raise HTTPException( status_code=403, detail=f'Permission {permission} required' ) return user return permission_checker

Level 5: Business logic (depends on authorization)

def get_user_service( db = Depends(get_db), cache = Depends(get_cache), current_user = Depends(get_active_user) ): return UserService(db=db, cache=cache, user=current_user)

Use in endpoint

@app.post('/users/{user_id}/deactivate') async def deactivate_user( user_id: int, admin = Depends(require_permission('users.deactivate')), service = Depends(get_user_service) ): result = await service.deactivate_user(user_id) return result

undefined

@app.post('/users/{user_id}/deactivate') async def deactivate_user( user_id: int, admin = Depends(require_permission('users.deactivate')), service = Depends(get_user_service) ): result = await service.deactivate_user(user_id) return result

undefined

OAuth2 Dependencies

OAuth2依赖项

Implement OAuth2 authentication with dependencies.

python

from fastapi import Depends, HTTPException, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    HTTPBearer,
    HTTPAuthorizationCredentials
)
from datetime import datetime, timedelta
from jose import JWTError, jwt

使用依赖项实现OAuth2认证。

python

from fastapi import Depends, HTTPException, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    HTTPBearer,
    HTTPAuthorizationCredentials
)
from datetime import datetime, timedelta
from jose import JWTError, jwt

OAuth2 with password flow

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

SECRET_KEY = 'your-secret-key-here' ALGORITHM = 'HS256' ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({'exp': expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt

async def get_current_user_from_token( token: str = Depends(oauth2_scheme), db = Depends(get_db) ): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail='Could not validate credentials', headers={'WWW-Authenticate': 'Bearer'}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) user_id: str = payload.get('sub') if user_id is None: raise credentials_exception except JWTError: raise credentials_exception

user = await db.get_user_by_id(user_id)
if user is None:
    raise credentials_exception
return user

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')

SECRET_KEY = 'your-secret-key-here' ALGORITHM = 'HS256' ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({'exp': expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt

async def get_current_user_from_token( token: str = Depends(oauth2_scheme), db = Depends(get_db) ): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail='Could not validate credentials', headers={'WWW-Authenticate': 'Bearer'}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) user_id: str = payload.get('sub') if user_id is None: raise credentials_exception except JWTError: raise credentials_exception

user = await db.get_user_by_id(user_id)
if user is None:
    raise credentials_exception
return user

OAuth2 with bearer token

bearer_scheme = HTTPBearer()

async def verify_bearer_token( credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme) ): token = credentials.credentials try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) return payload except JWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail='Invalid authentication credentials' )

bearer_scheme = HTTPBearer()

async def verify_bearer_token( credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme) ): token = credentials.credentials try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) return payload except JWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail='Invalid authentication credentials' )

OAuth2 with scopes

from fastapi.security import OAuth2PasswordBearer, SecurityScopes

oauth2_scheme_scopes = OAuth2PasswordBearer( tokenUrl='token', scopes={ 'users:read': 'Read user information', 'users:write': 'Modify user information', 'admin': 'Admin access' } )

async def get_current_user_with_scopes( security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme_scopes), db = Depends(get_db) ): if security_scopes.scopes: authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' else: authenticate_value = 'Bearer'

credentials_exception = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail='Could not validate credentials',
    headers={'WWW-Authenticate': authenticate_value},
)

try:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    user_id: str = payload.get('sub')
    if user_id is None:
        raise credentials_exception
    token_scopes = payload.get('scopes', [])
except JWTError:
    raise credentials_exception

user = await db.get_user_by_id(user_id)
if user is None:
    raise credentials_exception

for scope in security_scopes.scopes:
    if scope not in token_scopes:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail='Not enough permissions',
            headers={'WWW-Authenticate': authenticate_value},
        )
return user

from fastapi.security import OAuth2PasswordBearer, SecurityScopes

oauth2_scheme_scopes = OAuth2PasswordBearer( tokenUrl='token', scopes={ 'users:read': 'Read user information', 'users:write': 'Modify user information', 'admin': 'Admin access' } )

async def get_current_user_with_scopes( security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme_scopes), db = Depends(get_db) ): if security_scopes.scopes: authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' else: authenticate_value = 'Bearer'

credentials_exception = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail='Could not validate credentials',
    headers={'WWW-Authenticate': authenticate_value},
)

try:
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    user_id: str = payload.get('sub')
    if user_id is None:
        raise credentials_exception
    token_scopes = payload.get('scopes', [])
except JWTError:
    raise credentials_exception

user = await db.get_user_by_id(user_id)
if user is None:
    raise credentials_exception

for scope in security_scopes.scopes:
    if scope not in token_scopes:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail='Not enough permissions',
            headers={'WWW-Authenticate': authenticate_value},
        )
return user

Use with scopes

@app.get('/users/me', dependencies=[Security( get_current_user_with_scopes, scopes=['users:read'] )]) async def read_users_me( current_user = Depends(get_current_user_with_scopes) ): return current_user

@app.put('/users/me', dependencies=[Security( get_current_user_with_scopes, scopes=['users:write'] )]) async def update_user_me( user_update: UserUpdate, current_user = Depends(get_current_user_with_scopes), db = Depends(get_db) ): updated_user = await db.update_user(current_user.id, user_update) return updated_user

undefined

@app.get('/users/me', dependencies=[Security( get_current_user_with_scopes, scopes=['users:read'] )]) async def read_users_me( current_user = Depends(get_current_user_with_scopes) ): return current_user

@app.put('/users/me', dependencies=[Security( get_current_user_with_scopes, scopes=['users:write'] )]) async def update_user_me( user_update: UserUpdate, current_user = Depends(get_current_user_with_scopes), db = Depends(get_db) ): updated_user = await db.update_user(current_user.id, user_update) return updated_user

undefined

Database Dependencies

数据库依赖项

Manage database connections with dependencies.

python

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

使用依赖项管理数据库连接。

python

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

SQLAlchemy setup

DATABASE_URL = 'postgresql+asyncpg://user:pass@localhost/db' engine = create_async_engine(DATABASE_URL, echo=True) AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

Dependency

async def get_db() -> AsyncSession: async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close()

Usage

from sqlalchemy import select

@app.get('/users/{user_id}') async def get_user(user_id: int, db: AsyncSession = Depends(get_db)): result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail='User not found') return user

from sqlalchemy import select

@app.get('/users/{user_id}') async def get_user(user_id: int, db: AsyncSession = Depends(get_db)): result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail='User not found') return user

With transaction

@app.post('/users') async def create_user(user_data: UserCreate, db: AsyncSession = Depends(get_db)): user = User(**user_data.dict()) db.add(user) await db.flush() # Get the ID before commit return user

undefined

@app.post('/users') async def create_user(user_data: UserCreate, db: AsyncSession = Depends(get_db)): user = User(**user_data.dict()) db.add(user) await db.flush() # Get the ID before commit return user

undefined

Authentication Dependencies

认证依赖项

Implement authentication patterns.

python

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')

SECRET_KEY = 'your-secret-key'
ALGORITHM = 'HS256'

实现认证模式。

python

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext

oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token')
pwd_context = CryptContext(schemes=['bcrypt'], deprecated='auto')

SECRET_KEY = 'your-secret-key'
ALGORITHM = 'HS256'

Token verification

async def get_current_user( token: str = Depends(oauth2_scheme), db = Depends(get_db) ): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail='Could not validate credentials', headers={'WWW-Authenticate': 'Bearer'}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get('sub') if email is None: raise credentials_exception except JWTError: raise credentials_exception

user = await db.fetch_one('SELECT * FROM users WHERE email = ?', email)
if user is None:
    raise credentials_exception
return user

async def get_current_user( token: str = Depends(oauth2_scheme), db = Depends(get_db) ): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail='Could not validate credentials', headers={'WWW-Authenticate': 'Bearer'}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get('sub') if email is None: raise credentials_exception except JWTError: raise credentials_exception

user = await db.fetch_one('SELECT * FROM users WHERE email = ?', email)
if user is None:
    raise credentials_exception
return user

Active user check

async def get_current_active_user( current_user = Depends(get_current_user) ): if not current_user.is_active: raise HTTPException(status_code=400, detail='Inactive user') return current_user

Role-based access

def require_role(required_role: str): async def role_checker(current_user = Depends(get_current_active_user)): if current_user.role != required_role: raise HTTPException( status_code=403, detail=f'Role {required_role} required' ) return current_user return role_checker

@app.get('/admin') async def admin_route(user = Depends(require_role('admin'))): return {'message': 'Admin access granted'}

undefined

def require_role(required_role: str): async def role_checker(current_user = Depends(get_current_active_user)): if current_user.role != required_role: raise HTTPException( status_code=403, detail=f'Role {required_role} required' ) return current_user return role_checker

@app.get('/admin') async def admin_route(user = Depends(require_role('admin'))): return {'message': 'Admin access granted'}

undefined

Caching Dependencies

缓存依赖项

Implement caching patterns with dependencies.

python

from fastapi import Depends
from functools import lru_cache
import redis.asyncio as redis

使用依赖项实现缓存模式。

python

from fastapi import Depends
from functools import lru_cache
import redis.asyncio as redis

In-memory cache

@lru_cache() def get_settings(): return Settings()

@app.get('/config') async def get_config(settings: Settings = Depends(get_settings)): return settings

@lru_cache() def get_settings(): return Settings()

@app.get('/config') async def get_config(settings: Settings = Depends(get_settings)): return settings

Redis cache dependency

class RedisCache: def init(self): self.redis = None

async def get_connection(self):
    if not self.redis:
        self.redis = await redis.from_url('redis://localhost')
    return self.redis

async def get(self, key: str):
    conn = await self.get_connection()
    value = await conn.get(key)
    return value.decode() if value else None

async def set(self, key: str, value: str, expire: int = 3600):
    conn = await self.get_connection()
    await conn.set(key, value, ex=expire)

cache = RedisCache()

async def get_cache(): return cache

@app.get('/cached-data/{key}') async def get_cached_data( key: str, cache: RedisCache = Depends(get_cache) ): value = await cache.get(key) if value: return {'value': value, 'cached': True}

# Fetch and cache
value = fetch_expensive_data(key)
await cache.set(key, value)
return {'value': value, 'cached': False}

undefined

class RedisCache: def init(self): self.redis = None

async def get_connection(self):
    if not self.redis:
        self.redis = await redis.from_url('redis://localhost')
    return self.redis

async def get(self, key: str):
    conn = await self.get_connection()
    value = await conn.get(key)
    return value.decode() if value else None

async def set(self, key: str, value: str, expire: int = 3600):
    conn = await self.get_connection()
    await conn.set(key, value, ex=expire)

cache = RedisCache()

async def get_cache(): return cache

@app.get('/cached-data/{key}') async def get_cached_data( key: str, cache: RedisCache = Depends(get_cache) ): value = await cache.get(key) if value: return {'value': value, 'cached': True}

# Fetch and cache
value = fetch_expensive_data(key)
await cache.set(key, value)
return {'value': value, 'cached': False}

undefined

Background Task Dependencies

后台任务依赖项

Use dependencies with background tasks.

python

from fastapi import BackgroundTasks, Depends
import asyncio

async def send_email(email: str, message: str):
    # Simulate sending email
    await asyncio.sleep(2)
    print(f'Email sent to {email}: {message}')

def get_email_service():
    # Could return configured email service
    return send_email

@app.post('/users')
async def create_user(
    user: UserCreate,
    background_tasks: BackgroundTasks,
    email_service = Depends(get_email_service),
    db = Depends(get_db)
):
    user_obj = await db.create_user(user)
    background_tasks.add_task(
        email_service,
        user.email,
        'Welcome to our service!'
    )
    return user_obj

将依赖项与后台任务结合使用。

python

from fastapi import BackgroundTasks, Depends
import asyncio

async def send_email(email: str, message: str):
    # Simulate sending email
    await asyncio.sleep(2)
    print(f'Email sent to {email}: {message}')

def get_email_service():
    # Could return configured email service
    return send_email

@app.post('/users')
async def create_user(
    user: UserCreate,
    background_tasks: BackgroundTasks,
    email_service = Depends(get_email_service),
    db = Depends(get_db)
):
    user_obj = await db.create_user(user)
    background_tasks.add_task(
        email_service,
        user.email,
        'Welcome to our service!'
    )
    return user_obj

Complex background task with dependencies

class EmailService: def init(self, db = Depends(get_db)): self.db = db

async def send_welcome_email(self, user_id: int):
    user = await self.db.get_user(user_id)
    await send_email(user.email, f'Welcome {user.name}!')

@app.post('/users/v2') async def create_user_v2( user: UserCreate, background_tasks: BackgroundTasks, db = Depends(get_db) ): user_obj = await db.create_user(user) email_service = EmailService(db) background_tasks.add_task( email_service.send_welcome_email, user_obj.id ) return user_obj

class EmailService: def init(self, db = Depends(get_db)): self.db = db

async def send_welcome_email(self, user_id: int):
    user = await self.db.get_user(user_id)
    await send_email(user.email, f'Welcome {user.name}!')

@app.post('/users/v2') async def create_user_v2( user: UserCreate, background_tasks: BackgroundTasks, db = Depends(get_db) ): user_obj = await db.create_user(user) email_service = EmailService(db) background_tasks.add_task( email_service.send_welcome_email, user_obj.id ) return user_obj

Background task with cleanup

async def process_file_with_cleanup(file_path: str): try: # Process file await process_file(file_path) finally: # Cleanup os.remove(file_path)

@app.post('/upload') async def upload_file( file: UploadFile, background_tasks: BackgroundTasks ): file_path = f'/tmp/{file.filename}' with open(file_path, 'wb') as f: f.write(await file.read())

background_tasks.add_task(process_file_with_cleanup, file_path)
return {'status': 'processing'}

undefined

async def process_file_with_cleanup(file_path: str): try: # Process file await process_file(file_path) finally: # Cleanup os.remove(file_path)

@app.post('/upload') async def upload_file( file: UploadFile, background_tasks: BackgroundTasks ): file_path = f'/tmp/{file.filename}' with open(file_path, 'wb') as f: f.write(await file.read())

background_tasks.add_task(process_file_with_cleanup, file_path)
return {'status': 'processing'}

undefined

WebSocket Dependencies

WebSocket依赖项

Use dependencies with WebSocket connections.

python

from fastapi import WebSocket, Depends, WebSocketDisconnect
from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

def get_connection_manager():
    return manager

将依赖项与WebSocket连接结合使用。

python

from fastapi import WebSocket, Depends, WebSocketDisconnect
from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

def get_connection_manager():
    return manager

WebSocket with authentication

async def get_ws_current_user( websocket: WebSocket, token: str = Query(...), db = Depends(get_db) ): try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) user_id = payload.get('sub') user = await db.get_user_by_id(user_id) if not user: await websocket.close(code=1008) raise HTTPException(status_code=401, detail='Invalid token') return user except JWTError: await websocket.close(code=1008) raise HTTPException(status_code=401, detail='Invalid token')

@app.websocket('/ws/{client_id}') async def websocket_endpoint( websocket: WebSocket, client_id: int, manager: ConnectionManager = Depends(get_connection_manager), user = Depends(get_ws_current_user) ): await manager.connect(websocket) try: while True: data = await websocket.receive_text() message = f'User {user.name}: {data}' await manager.broadcast(message) except WebSocketDisconnect: manager.disconnect(websocket) await manager.broadcast(f'User {user.name} left the chat')

async def get_ws_current_user( websocket: WebSocket, token: str = Query(...), db = Depends(get_db) ): try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) user_id = payload.get('sub') user = await db.get_user_by_id(user_id) if not user: await websocket.close(code=1008) raise HTTPException(status_code=401, detail='Invalid token') return user except JWTError: await websocket.close(code=1008) raise HTTPException(status_code=401, detail='Invalid token')

@app.websocket('/ws/{client_id}') async def websocket_endpoint( websocket: WebSocket, client_id: int, manager: ConnectionManager = Depends(get_connection_manager), user = Depends(get_ws_current_user) ): await manager.connect(websocket) try: while True: data = await websocket.receive_text() message = f'User {user.name}: {data}' await manager.broadcast(message) except WebSocketDisconnect: manager.disconnect(websocket) await manager.broadcast(f'User {user.name} left the chat')

WebSocket with database

@app.websocket('/ws/messages/{room_id}') async def message_websocket( websocket: WebSocket, room_id: int, db = Depends(get_db), user = Depends(get_ws_current_user) ): await websocket.accept()

# Send message history
messages = await db.get_room_messages(room_id)
await websocket.send_json(messages)

try:
    while True:
        data = await websocket.receive_text()
        # Save to database
        message = await db.create_message(
            room_id=room_id,
            user_id=user.id,
            content=data
        )
        await websocket.send_json(message)
except WebSocketDisconnect:
    pass

undefined

@app.websocket('/ws/messages/{room_id}') async def message_websocket( websocket: WebSocket, room_id: int, db = Depends(get_db), user = Depends(get_ws_current_user) ): await websocket.accept()

# Send message history
messages = await db.get_room_messages(room_id)
await websocket.send_json(messages)

try:
    while True:
        data = await websocket.receive_text()
        # Save to database
        message = await db.create_message(
            room_id=room_id,
            user_id=user.id,
            content=data
        )
        await websocket.send_json(message)
except WebSocketDisconnect:
    pass

undefined

Custom Dependency Providers

自定义依赖项提供器

Create custom dependency injection patterns.

python

from typing import Callable, Type, TypeVar, Generic
from fastapi import Depends

T = TypeVar('T')

创建自定义依赖注入模式。

python

from typing import Callable, Type, TypeVar, Generic
from fastapi import Depends

T = TypeVar('T')

Dependency factory

class DependencyFactory(Generic[T]): def init(self, dependency_class: Type[T], **kwargs): self.dependency_class = dependency_class self.kwargs = kwargs

def __call__(self) -> T:
    return self.dependency_class(**self.kwargs)

class DependencyFactory(Generic[T]): def init(self, dependency_class: Type[T], **kwargs): self.dependency_class = dependency_class self.kwargs = kwargs

def __call__(self) -> T:
    return self.dependency_class(**self.kwargs)

Service locator pattern

class ServiceLocator: def init(self): self._services = {}

def register(self, name: str, service):
    self._services[name] = service

def get(self, name: str):
    return self._services.get(name)

def create_dependency(self, name: str):
    def get_service():
        service = self.get(name)
        if service is None:
            raise ValueError(f'Service {name} not registered')
        return service
    return get_service

class ServiceLocator: def init(self): self._services = {}

def register(self, name: str, service):
    self._services[name] = service

def get(self, name: str):
    return self._services.get(name)

def create_dependency(self, name: str):
    def get_service():
        service = self.get(name)
        if service is None:
            raise ValueError(f'Service {name} not registered')
        return service
    return get_service

Initialize service locator

locator = ServiceLocator() locator.register('db', Database('postgresql://localhost/db')) locator.register('cache', Redis('redis://localhost'))

Use in endpoint

get_db_from_locator = locator.create_dependency('db')

@app.get('/items') async def get_items(db = Depends(get_db_from_locator)): return await db.fetch_all('SELECT * FROM items')

get_db_from_locator = locator.create_dependency('db')

@app.get('/items') async def get_items(db = Depends(get_db_from_locator)): return await db.fetch_all('SELECT * FROM items')

Dependency provider with context

class ContextualDependency: def init(self, request: Request): self.request = request self.context = {}

def set(self, key: str, value):
    self.context[key] = value

def get(self, key: str):
    return self.context.get(key)

async def get_request_context(request: Request): context = ContextualDependency(request) context.set('request_id', str(uuid.uuid4())) context.set('timestamp', datetime.utcnow()) return context

@app.get('/context-example') async def context_example(context = Depends(get_request_context)): return { 'request_id': context.get('request_id'), 'timestamp': context.get('timestamp') }

class ContextualDependency: def init(self, request: Request): self.request = request self.context = {}

def set(self, key: str, value):
    self.context[key] = value

def get(self, key: str):
    return self.context.get(key)

async def get_request_context(request: Request): context = ContextualDependency(request) context.set('request_id', str(uuid.uuid4())) context.set('timestamp', datetime.utcnow()) return context

@app.get('/context-example') async def context_example(context = Depends(get_request_context)): return { 'request_id': context.get('request_id'), 'timestamp': context.get('timestamp') }

Lazy dependency loading

class LazyDependency: def init(self, factory: Callable): self.factory = factory self._instance = None

def __call__(self):
    if self._instance is None:
        self._instance = self.factory()
    return self._instance

def create_expensive_service(): print('Creating expensive service...') return ExpensiveService()

expensive_service = LazyDependency(create_expensive_service)

@app.get('/expensive') async def use_expensive(service = Depends(expensive_service)): return service.do_work()

undefined

class LazyDependency: def init(self, factory: Callable): self.factory = factory self._instance = None

def __call__(self):
    if self._instance is None:
        self._instance = self.factory()
    return self._instance

def create_expensive_service(): print('Creating expensive service...') return ExpensiveService()

expensive_service = LazyDependency(create_expensive_service)

@app.get('/expensive') async def use_expensive(service = Depends(expensive_service)): return service.do_work()

undefined

Scoped Dependencies Per Request

每个请求的作用域依赖项

Manage request-scoped state and lifecycle.

python

from contextvars import ContextVar
from fastapi import Depends, Request

管理请求作用域的状态与生命周期。

python

from contextvars import ContextVar
from fastapi import Depends, Request

Request-scoped storage using context variables

request_id_var: ContextVar[str] = ContextVar('request_id')

async def set_request_id(request: Request): request_id = request.headers.get('X-Request-ID', str(uuid.uuid4())) request_id_var.set(request_id) return request_id

async def get_request_id(): return request_id_var.get()

request_id_var: ContextVar[str] = ContextVar('request_id')

async def set_request_id(request: Request): request_id = request.headers.get('X-Request-ID', str(uuid.uuid4())) request_id_var.set(request_id) return request_id

async def get_request_id(): return request_id_var.get()

Request-scoped logger

class RequestLogger: def init(self, request_id: str = Depends(get_request_id)): self.request_id = request_id

def info(self, message: str):
    print(f'[{self.request_id}] INFO: {message}')

def error(self, message: str):
    print(f'[{self.request_id}] ERROR: {message}')

@app.get('/scoped-logging') async def scoped_logging( request_id: str = Depends(set_request_id), logger: RequestLogger = Depends() ): logger.info('Processing request') # Do work logger.info('Request completed') return {'request_id': request_id}

class RequestLogger: def init(self, request_id: str = Depends(get_request_id)): self.request_id = request_id

def info(self, message: str):
    print(f'[{self.request_id}] INFO: {message}')

def error(self, message: str):
    print(f'[{self.request_id}] ERROR: {message}')

@app.get('/scoped-logging') async def scoped_logging( request_id: str = Depends(set_request_id), logger: RequestLogger = Depends() ): logger.info('Processing request') # Do work logger.info('Request completed') return {'request_id': request_id}

Request-scoped unit of work pattern

class UnitOfWork: def init(self, db = Depends(get_db)): self.db = db self.repositories = {}

def get_repository(self, model_class):
    if model_class not in self.repositories:
        self.repositories[model_class] = Repository(self.db, model_class)
    return self.repositories[model_class]

async def commit(self):
    await self.db.commit()

async def rollback(self):
    await self.db.rollback()

@app.post('/complex-transaction') async def complex_transaction( data: TransactionData, uow: UnitOfWork = Depends() ): try: user_repo = uow.get_repository(User) order_repo = uow.get_repository(Order)

    user = await user_repo.create(data.user)
    order = await order_repo.create(data.order, user_id=user.id)

    await uow.commit()
    return {'user': user, 'order': order}
except Exception:
    await uow.rollback()
    raise

undefined

class UnitOfWork: def init(self, db = Depends(get_db)): self.db = db self.repositories = {}

def get_repository(self, model_class):
    if model_class not in self.repositories:
        self.repositories[model_class] = Repository(self.db, model_class)
    return self.repositories[model_class]

async def commit(self):
    await self.db.commit()

async def rollback(self):
    await self.db.rollback()

@app.post('/complex-transaction') async def complex_transaction( data: TransactionData, uow: UnitOfWork = Depends() ): try: user_repo = uow.get_repository(User) order_repo = uow.get_repository(Order)

    user = await user_repo.create(data.user)
    order = await order_repo.create(data.order, user_id=user.id)

    await uow.commit()
    return {'user': user, 'order': order}
except Exception:
    await uow.rollback()
    raise

undefined

Global Dependencies with app.dependency_overrides

使用app.dependency_overrides的全局依赖项

Use global dependency management and overrides.

python

from fastapi import FastAPI, Depends

app = FastAPI()

使用全局依赖项管理与覆盖机制。

python

from fastapi import FastAPI, Depends

app = FastAPI()

Original dependencies

def get_production_db(): return ProductionDatabase('postgresql://prod/db')

def get_production_cache(): return RedisCache('redis://prod')

def get_production_db(): return ProductionDatabase('postgresql://prod/db')

def get_production_cache(): return RedisCache('redis://prod')

Default app setup

@app.get('/data') async def get_data( db = Depends(get_production_db), cache = Depends(get_production_cache) ): cached = await cache.get('data') if cached: return cached

data = await db.fetch_all('SELECT * FROM data')
await cache.set('data', data)
return data

@app.get('/data') async def get_data( db = Depends(get_production_db), cache = Depends(get_production_cache) ): cached = await cache.get('data') if cached: return cached

data = await db.fetch_all('SELECT * FROM data')
await cache.set('data', data)
return data

Override for testing environment

if os.getenv('ENVIRONMENT') == 'test': def get_test_db(): return TestDatabase(':memory:')

def get_test_cache():
    return InMemoryCache()

app.dependency_overrides[get_production_db] = get_test_db
app.dependency_overrides[get_production_cache] = get_test_cache

if os.getenv('ENVIRONMENT') == 'test': def get_test_db(): return TestDatabase(':memory:')

def get_test_cache():
    return InMemoryCache()

app.dependency_overrides[get_production_db] = get_test_db
app.dependency_overrides[get_production_cache] = get_test_cache

Override for development

if os.getenv('ENVIRONMENT') == 'development': def get_dev_db(): return DevDatabase('postgresql://localhost/dev')

app.dependency_overrides[get_production_db] = get_dev_db

if os.getenv('ENVIRONMENT') == 'development': def get_dev_db(): return DevDatabase('postgresql://localhost/dev')

app.dependency_overrides[get_production_db] = get_dev_db

Dynamic override based on request

async def override_db_by_tenant(request: Request): tenant_id = request.headers.get('X-Tenant-ID') if tenant_id: return TenantDatabase(tenant_id) return get_production_db()

Conditional override

def setup_dependencies(app: FastAPI, config: dict): if config.get('use_mock_db'): app.dependency_overrides[get_production_db] = lambda: MockDB()

if config.get('use_local_cache'):
    app.dependency_overrides[get_production_cache] = lambda: LocalCache()

def setup_dependencies(app: FastAPI, config: dict): if config.get('use_mock_db'): app.dependency_overrides[get_production_db] = lambda: MockDB()

if config.get('use_local_cache'):
    app.dependency_overrides[get_production_cache] = lambda: LocalCache()

Clear overrides

def clear_overrides(): app.dependency_overrides = {}

undefined

def clear_overrides(): app.dependency_overrides = {}

undefined

Testing with Dependencies

依赖项测试

Override dependencies for testing.

python

from fastapi.testclient import TestClient

覆盖依赖项以进行测试。

python

from fastapi.testclient import TestClient

Original dependency

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

app.dependency_overrides = {}

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

app.dependency_overrides = {}

Test with mock database

def test_get_users(): def override_get_db(): return MockDB()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)
response = client.get('/users')

assert response.status_code == 200

# Cleanup
app.dependency_overrides = {}

def test_get_users(): def override_get_db(): return MockDB()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)
response = client.get('/users')

assert response.status_code == 200

# Cleanup
app.dependency_overrides = {}

Pytest fixture for dependency override

import pytest

@pytest.fixture def client(): def override_get_db(): return MockDB()

app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
yield client
app.dependency_overrides = {}

def test_with_fixture(client): response = client.get('/users') assert response.status_code == 200

undefined

import pytest

@pytest.fixture def client(): def override_get_db(): return MockDB()

app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
yield client
app.dependency_overrides = {}

def test_with_fixture(client): response = client.get('/users') assert response.status_code == 200

undefined

Global Dependencies

全局依赖项

Apply dependencies to all routes.

python

from fastapi import FastAPI, Depends

将依赖项应用于所有路由。

python

from fastapi import FastAPI, Depends

Logging dependency

async def log_request(request: Request): print(f'{request.method} {request.url}')

Rate limiting

async def rate_limit(request: Request): client_ip = request.client.host # Check rate limit if is_rate_limited(client_ip): raise HTTPException(status_code=429, detail='Too many requests')

Apply globally

app = FastAPI(dependencies=[Depends(log_request), Depends(rate_limit)])

Apply to router

router = APIRouter(dependencies=[Depends(get_current_user)])

@router.get('/protected-resource') async def protected_route(): return {'message': 'This requires authentication'}

app.include_router(router)

undefined

router = APIRouter(dependencies=[Depends(get_current_user)])

@router.get('/protected-resource') async def protected_route(): return {'message': 'This requires authentication'}

app.include_router(router)

undefined

When to Use This Skill

何时使用该技能

Use fastapi-dependency-injection when building modern, production-ready applications that require advanced patterns, best practices, and optimal performance.

在构建需要高级模式、最佳实践与最优性能的现代生产级应用时，使用fastapi-dependency-injection。

FastAPI DI Best Practices

FastAPI依赖注入最佳实践

Use yield for cleanup - Always use yield for resources that need cleanup like database connections, file handles, and network connections to ensure proper resource management
Leverage caching - Use dependency caching (enabled by default) to avoid redundant work within a request; multiple uses of the same dependency in one request share the same instance
Chain dependencies - Build complex dependencies from simpler sub-dependencies to create composable, testable, and maintainable code structures
Class-based for state - Use classes for dependencies that maintain state or configuration, leveraging
```
__call__
```
method for callable instances
Type hints everywhere - Always add type hints for better editor support, automatic validation, and improved documentation generation
Override for testing - Use dependency_overrides to inject mocks during testing without modifying production code
Global for cross-cutting - Apply common dependencies (logging, auth, rate limiting) globally or at router level to avoid repetition
Async when possible - Use async dependencies for I/O operations to maximize performance and concurrency benefits
Separate concerns - Keep authentication, authorization, and business logic in separate dependencies for better testability and reusability
Document dependencies - Add docstrings to explain complex dependency chains, especially when building multi-level hierarchies
Use Security utilities - Leverage FastAPI's security utilities like OAuth2PasswordBearer and HTTPBearer for authentication patterns
Validate early - Place validation dependencies early in the chain to fail fast and provide clear error messages
Keep dependencies pure - Dependencies should have minimal side effects; use background tasks for non-critical operations
Use context managers - Wrap dependencies in context managers when dealing with transactions or resource pools
Dependency composition - Compose larger dependencies from smaller, focused ones rather than creating monolithic dependencies

使用yield进行清理 - 对于需要清理的资源（如数据库连接、文件句柄、网络连接），始终使用yield以确保资源被正确管理
利用缓存 - 使用依赖项缓存（默认启用）避免请求内的重复工作；同一请求中多次使用同一依赖项会共享同一实例
链式依赖项 - 从简单的子依赖项构建复杂依赖项，创建可组合、可测试、可维护的代码结构
基于类实现状态 - 对于需要维护状态或配置的依赖项，使用类并利用
```
__call__
```
方法实现可调用实例
处处使用类型提示 - 始终添加类型提示以获得更好的编辑器支持、自动验证与更完善的文档生成
测试时覆盖依赖项 - 使用dependency_overrides在测试时注入模拟对象，无需修改生产代码
全局应用横切关注点 - 将通用依赖项（日志、认证、限流）全局或路由级应用，避免重复代码
尽可能使用异步 - 对I/O操作使用异步依赖项以最大化性能与并发优势
分离关注点 - 将认证、授权与业务逻辑放在单独的依赖项中，提升可测试性与可复用性
文档化依赖项 - 为复杂的依赖项链添加文档字符串，尤其是在构建多层级结构时
使用安全工具 - 利用FastAPI的安全工具（如OAuth2PasswordBearer、HTTPBearer）实现认证模式
尽早验证 - 将验证依赖项放在链的早期，快速失败并提供清晰的错误信息
保持依赖项纯净 - 依赖项应尽量减少副作用；对于非关键操作使用后台任务
使用上下文管理器 - 处理事务或资源池时，将依赖项包装在上下文管理器中
依赖项组合 - 从更小、聚焦的依赖项组合出更大的依赖项，而非创建单体式依赖项

FastAPI DI Common Pitfalls

FastAPI依赖注入常见陷阱

Forgetting yield - Not using yield means resources won't be cleaned up properly, leading to connection leaks and resource exhaustion
Circular dependencies - Creating dependency cycles causes infinite loops and stack overflow errors; design dependencies in a directed acyclic graph
Not using Depends() - Forgetting Depends() wrapper means function is called directly instead of being injected, breaking the dependency resolution
Overusing use_cache=False - Disabling cache unnecessarily hurts performance by creating multiple instances of the same dependency per request
Heavy dependencies - Putting too much logic in dependencies instead of services makes them hard to test and violates single responsibility
Not testing overrides - Forgetting to test with dependency_overrides means tests may use production resources instead of mocks
Mixing sync and async - Incorrectly mixing synchronous and asynchronous dependencies can block the event loop or cause runtime errors
Global state issues - Not properly managing singleton dependencies leads to shared state bugs in concurrent requests
Exception handling - Not handling exceptions in dependencies properly can leave resources in inconsistent states or leak connections
Type hint mistakes - Missing or incorrect type hints break dependency injection and automatic validation
Ignoring dependency order - Dependencies are executed in the order they appear in function signature; incorrect order can cause issues
Not cleaning test overrides - Forgetting to reset app.dependency_overrides after tests causes subsequent tests to fail
Overusing global dependencies - Applying too many dependencies globally can hurt performance and make debugging difficult
Memory leaks with generators - Not properly closing resources in finally block of generator dependencies causes memory leaks
Security misconfiguration - Using weak or missing security dependencies exposes endpoints to unauthorized access

忘记使用yield - 不使用yield会导致资源无法被正确清理，引发连接泄漏与资源耗尽
循环依赖项 - 创建依赖项循环会导致无限循环与栈溢出错误；将依赖项设计为有向无环图
未使用Depends() - 忘记Depends()包装器会导致函数被直接调用而非注入，破坏依赖项解析
过度使用use_cache=False - 不必要地禁用缓存会因同一请求中创建多个同一依赖项实例而损害性能
过重的依赖项 - 在依赖项中放入过多逻辑而非服务中，会使其难以测试并违反单一职责原则
未测试覆盖项 - 忘记使用dependency_overrides测试会导致测试使用生产资源而非模拟对象
混合同步与异步 - 错误地混合同步与异步依赖项会阻塞事件循环或导致运行时错误
全局状态问题 - 未正确管理单例依赖项会导致并发请求中的共享状态bug
异常处理不当 - 未正确处理依赖项中的异常会导致资源处于不一致状态或连接泄漏
类型提示错误 - 缺失或错误的类型提示会破坏依赖注入与自动验证
忽略依赖项顺序 - 依赖项按函数签名中的顺序执行；错误的顺序会引发问题
未清理测试覆盖项 - 测试后忘记重置app.dependency_overrides会导致后续测试失败
过度使用全局依赖项 - 应用过多全局依赖项会损害性能并增加调试难度
生成器导致内存泄漏 - 未在生成器依赖项的finally块中正确关闭资源会导致内存泄漏
安全配置错误 - 使用弱或缺失的安全依赖项会使端点暴露于未授权访问风险

Advanced Caching Patterns

高级缓存模式

Implement sophisticated caching strategies with dependencies.

python

from fastapi import Depends
import hashlib
import json

使用依赖项实现复杂的缓存策略。

python

from fastapi import Depends
import hashlib
import json

Multi-layer cache with fallback

class CacheLayer: def init( self, memory_cache = Depends(get_memory_cache), redis_cache = Depends(get_redis_cache) ): self.memory = memory_cache self.redis = redis_cache

async def get(self, key: str):
    # Try memory first
    value = self.memory.get(key)
    if value:
        return value

    # Try Redis
    value = await self.redis.get(key)
    if value:
        # Populate memory cache
        self.memory.set(key, value)
        return value

    return None

async def set(self, key: str, value, ttl: int = 3600):
    self.memory.set(key, value, ttl=min(ttl, 300))
    await self.redis.set(key, value, ttl=ttl)

class CacheLayer: def init( self, memory_cache = Depends(get_memory_cache), redis_cache = Depends(get_redis_cache) ): self.memory = memory_cache self.redis = redis_cache

async def get(self, key: str):
    # Try memory first
    value = self.memory.get(key)
    if value:
        return value

    # Try Redis
    value = await self.redis.get(key)
    if value:
        # Populate memory cache
        self.memory.set(key, value)
        return value

    return None

async def set(self, key: str, value, ttl: int = 3600):
    self.memory.set(key, value, ttl=min(ttl, 300))
    await self.redis.set(key, value, ttl=ttl)

Cache key generation

def create_cache_key(*args, **kwargs): key_data = json.dumps({'args': args, 'kwargs': kwargs}, sort_keys=True) return hashlib.md5(key_data.encode()).hexdigest()

Dependency with automatic caching

def cached_dependency(ttl: int = 3600): async def dependency( params: dict, cache: CacheLayer = Depends() ): cache_key = create_cache_key(**params) cached_value = await cache.get(cache_key)

    if cached_value:
        return cached_value

    # Compute expensive value
    value = await compute_expensive_value(params)
    await cache.set(cache_key, value, ttl=ttl)
    return value

return dependency

@app.get('/cached-endpoint') async def cached_endpoint( result = Depends(cached_dependency(ttl=1800)) ): return result

def cached_dependency(ttl: int = 3600): async def dependency( params: dict, cache: CacheLayer = Depends() ): cache_key = create_cache_key(**params) cached_value = await cache.get(cache_key)

    if cached_value:
        return cached_value

    # Compute expensive value
    value = await compute_expensive_value(params)
    await cache.set(cache_key, value, ttl=ttl)
    return value

return dependency

@app.get('/cached-endpoint') async def cached_endpoint( result = Depends(cached_dependency(ttl=1800)) ): return result

Cache invalidation dependency

class CacheInvalidator: def init(self, cache: CacheLayer = Depends()): self.cache = cache self.invalidation_queue = []

def invalidate(self, pattern: str):
    self.invalidation_queue.append(pattern)

async def flush(self):
    for pattern in self.invalidation_queue:
        await self.cache.redis.delete_pattern(pattern)
    self.invalidation_queue.clear()

@app.post('/users') async def create_user( user: UserCreate, db = Depends(get_db), invalidator: CacheInvalidator = Depends() ): new_user = await db.create_user(user) invalidator.invalidate('users:*') await invalidator.flush() return new_user

undefined

class CacheInvalidator: def init(self, cache: CacheLayer = Depends()): self.cache = cache self.invalidation_queue = []

def invalidate(self, pattern: str):
    self.invalidation_queue.append(pattern)

async def flush(self):
    for pattern in self.invalidation_queue:
        await self.cache.redis.delete_pattern(pattern)
    self.invalidation_queue.clear()

@app.post('/users') async def create_user( user: UserCreate, db = Depends(get_db), invalidator: CacheInvalidator = Depends() ): new_user = await db.create_user(user) invalidator.invalidate('users:*') await invalidator.flush() return new_user

undefined

Middleware-Style Dependencies

中间件风格的依赖项

Use dependencies for cross-cutting concerns.

python

from fastapi import Depends, Request
from time import time

使用依赖项处理横切关注点。

python

from fastapi import Depends, Request
from time import time

Request timing dependency

async def measure_request_time(request: Request): start_time = time() yield duration = time() - start_time print(f'{request.method} {request.url.path} took {duration:.3f}s')

Request ID tracking

async def track_request_id(request: Request): request_id = request.headers.get('X-Request-ID', str(uuid.uuid4())) request.state.request_id = request_id yield request_id

def get_request_id(request: Request): return request.state.request_id

async def track_request_id(request: Request): request_id = request.headers.get('X-Request-ID', str(uuid.uuid4())) request.state.request_id = request_id yield request_id

def get_request_id(request: Request): return request.state.request_id

Rate limiting per user

class RateLimiter: def init(self): self.requests = {}

async def check_rate_limit(
    self,
    user = Depends(get_current_user),
    cache = Depends(get_cache)
):
    key = f'rate_limit:{user.id}'
    count = await cache.incr(key)

    if count == 1:
        await cache.expire(key, 60)

    if count > 100:
        raise HTTPException(
            status_code=429,
            detail='Rate limit exceeded'
        )
    return True

rate_limiter = RateLimiter()

@app.get('/protected') async def protected_endpoint( rate_limit_ok = Depends(rate_limiter.check_rate_limit) ): return {'message': 'Success'}

class RateLimiter: def init(self): self.requests = {}

async def check_rate_limit(
    self,
    user = Depends(get_current_user),
    cache = Depends(get_cache)
):
    key = f'rate_limit:{user.id}'
    count = await cache.incr(key)

    if count == 1:
        await cache.expire(key, 60)

    if count > 100:
        raise HTTPException(
            status_code=429,
            detail='Rate limit exceeded'
        )
    return True

rate_limiter = RateLimiter()

@app.get('/protected') async def protected_endpoint( rate_limit_ok = Depends(rate_limiter.check_rate_limit) ): return {'message': 'Success'}

Request validation

async def validate_content_type(request: Request): content_type = request.headers.get('Content-Type') if not content_type or 'application/json' not in content_type: raise HTTPException( status_code=415, detail='Content-Type must be application/json' ) return True

@app.post('/data', dependencies=[Depends(validate_content_type)]) async def post_data(data: dict): return data

undefined

async def validate_content_type(request: Request): content_type = request.headers.get('Content-Type') if not content_type or 'application/json' not in content_type: raise HTTPException( status_code=415, detail='Content-Type must be application/json' ) return True

@app.post('/data', dependencies=[Depends(validate_content_type)]) async def post_data(data: dict): return data

undefined

Resources

资源

Official FastAPI documentation and guides:

FastAPI Dependencies Core dependency injection concepts and basic usage
Advanced Dependencies Advanced patterns including dependency overrides and custom providers
Sub-Dependencies Building complex dependency chains and hierarchies
Dependencies in Path Operations Using dependencies at the decorator level
Global Dependencies Applying dependencies to entire applications or routers
Testing Dependencies Overriding dependencies for testing purposes
Security and OAuth2 Security utilities and OAuth2 implementation with dependencies
SQL Databases Database session management with dependency injection
Async SQL Async database patterns with SQLAlchemy

Additional resources:

FastAPI GitHub Source code and examples
FastAPI Discussions Community questions and patterns
Dependency Injection Design Pattern General DI concepts and theory

官方FastAPI文档与指南：

FastAPI Dependencies 核心依赖注入概念与基础用法
Advanced Dependencies 高级模式，包括依赖项覆盖与自定义提供器
Sub-Dependencies 构建复杂依赖项链与层级结构
Dependencies in Path Operations 在装饰器级别使用依赖项
Global Dependencies 将依赖项应用于整个应用或路由
Testing Dependencies 覆盖依赖项以进行测试
Security and OAuth2 安全工具与基于依赖项的OAuth2实现
SQL Databases 使用依赖注入管理数据库会话
Async SQL 结合SQLAlchemy的异步数据库模式

额外资源：

FastAPI GitHub 源代码与示例
FastAPI Discussions 社区问题与模式
Dependency Injection Design Pattern 通用依赖注入概念与理论

fastapi-dependency-injection

Original

Translation

FastAPI Dependency Injection

FastAPI依赖注入

Basic Dependencies

基础依赖项

Simple function dependency

Simple function dependency

Async dependencies

Async dependencies

Dependency Scopes

依赖项作用域

Request-scoped dependency (default, cached per request)

Request-scoped dependency (default, cached per request)

Multiple uses in same request reuse the same instance

Multiple uses in same request reuse the same instance

No caching (use_cache=False)

No caching (use_cache=False)

Singleton pattern (application scope)

Singleton pattern (application scope)

Sub-Dependencies

子依赖项

Sub-dependency chain

Sub-dependency chain

Use in endpoint

Use in endpoint

Class-Based Dependencies

基于类的依赖项

Callable class (using call)

Callable class (using call)

Class with initialization parameters

Class with initialization parameters

Advanced class-based dependency with configuration

Advanced class-based dependency with configuration

Factory function for configurable class-based dependency

Factory function for configurable class-based dependency

Generator Dependencies for Cleanup

用于资源清理的生成器依赖项

Database connection with cleanup

Database connection with cleanup

HTTP client with cleanup

HTTP client with cleanup

File handle with cleanup

File handle with cleanup

Multiple resources with cleanup

Multiple resources with cleanup

Context manager as dependency

Context manager as dependency

Dependency Chains and Sub-Dependencies

依赖项链与子依赖项

Level 1: Configuration

Level 1: Configuration

Level 2: Infrastructure (depends on config)

Level 2: Infrastructure (depends on config)

Level 3: Authentication (depends on infrastructure)

Level 3: Authentication (depends on infrastructure)

Level 4: Authorization (depends on authentication)

Level 4: Authorization (depends on authentication)

Level 5: Business logic (depends on authorization)

Level 5: Business logic (depends on authorization)

Use in endpoint

Use in endpoint

OAuth2 Dependencies

OAuth2依赖项

OAuth2 with password flow

OAuth2 with password flow

OAuth2 with bearer token

OAuth2 with bearer token

OAuth2 with scopes

OAuth2 with scopes

Use with scopes

Use with scopes

Database Dependencies

数据库依赖项

SQLAlchemy setup

SQLAlchemy setup

Dependency

Dependency

Usage