Dependency Injection trong FastAPI: Hướng Dẫn Toàn Diện

# ❌ Without Dependency Injection
def get_user(user_id: int):
    db = Database()  # Creating dependency internally
    return db.query(User).filter(User.id == user_id).first()
 
# ✅ With Dependency Injection
def get_user(user_id: int, db: Database = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

from fastapi import Depends, FastAPI
 
app = FastAPI()
 
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    """Common query parameters."""
    return {"q": q, "skip": skip, "limit": limit}
 
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    """Use common parameters dependency."""
    return commons
 
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    """Reuse the same dependency."""
    return commons

from sqlalchemy.orm import Session
from database import SessionLocal
 
def get_db():
    """Database session dependency."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    """Get user with automatic DB session management."""
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

from typing import Optional
 
class CommonQueryParams:
    """Reusable query parameters as a class."""
    
    def __init__(
        self,
        q: Optional[str] = None,
        skip: int = 0,
        limit: int = 100
    ):
        self.q = q
        self.skip = skip
        self.limit = limit
 
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    """Use class-based dependency."""
    response = {"skip": commons.skip, "limit": commons.limit}
    if commons.q:
        response["q"] = commons.q
    return response
 
# Shorthand syntax (same result)
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
    """FastAPI infers the dependency from the type hint."""
    return commons

from fastapi import Header, HTTPException
 
def verify_token(x_token: str = Header(...)):
    """Verify authentication token."""
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    return x_token
 
def verify_key(x_key: str = Header(...)):
    """Verify API key."""
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="Invalid X-Key header")
    return x_key
 
def get_current_user(
    token: str = Depends(verify_token),
    key: str = Depends(verify_key),
    db: Session = Depends(get_db)
):
    """
    Get current user - depends on verify_token, verify_key, and get_db.
    FastAPI resolves all sub-dependencies automatically.
    """
    # Both token and key are verified before this runs
    user = db.query(User).filter(User.token == token).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user
 
@app.get("/users/me")
async def read_user_me(current_user: User = Depends(get_current_user)):
    """
    This endpoint requires:
    1. Valid X-Token header (via verify_token)
    2. Valid X-Key header (via verify_key)
    3. Database session (via get_db)
    4. User exists in database (via get_current_user)
    
    All checked automatically by FastAPI!
    """
    return current_user

read_user_me
    └── get_current_user
            ├── verify_token
            ├── verify_key
            └── get_db

from fastapi import Depends
 
# Apply dependency to all routes in the decorator
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    """This route requires token and key, but doesn't need their return values."""
    return [{"item_id": "Foo"}]
 
# Multiple dependencies
@app.get(
    "/users/",
    dependencies=[
        Depends(verify_token),
        Depends(verify_key),
        Depends(check_subscription)
    ]
)
async def read_users():
    """Multiple dependencies for access control."""
    return [{"username": "Rick"}, {"username": "Morty"}]

from fastapi import FastAPI, Depends
 
app = FastAPI(
    dependencies=[Depends(verify_token)]
)
 
# Now ALL routes require the token
@app.get("/items/")
async def read_items():
    return [{"item_id": "Foo"}]
 
@app.get("/users/")
async def read_users():
    return [{"username": "Rick"}]

from fastapi import APIRouter
 
router = APIRouter(
    prefix="/admin",
    dependencies=[Depends(verify_admin)]
)
 
@router.get("/users")
async def admin_get_users():
    """Only admins can access this."""
    return users
 
@router.delete("/users/{user_id}")
async def admin_delete_user(user_id: int):
    """Only admins can access this."""
    return {"deleted": user_id}

from sqlalchemy.orm import Session
 
def get_db_with_transaction():
    """Database with automatic transaction management."""
    db = SessionLocal()
    try:
        yield db
        db.commit()  # Commit on success
    except Exception:
        db.rollback()  # Rollback on error
        raise
    finally:
        db.close()  # Always close
 
@app.post("/users/")
async def create_user(
    user: UserCreate,
    db: Session = Depends(get_db_with_transaction)
):
    """User is created in a transaction."""
    new_user = User(**user.dict())
    db.add(new_user)
    # Transaction commits automatically if no exception
    return new_user

import time
from contextlib import contextmanager
 
class Timer:
    """Timer context manager for performance monitoring."""
    
    def __enter__(self):
        self.start = time.time()
        return self
    
    def __exit__(self, *args):
        self.end = time.time()
        self.elapsed = self.end - self.start
 
def get_timer():
    """Timer dependency with automatic cleanup."""
    timer = Timer()
    try:
        with timer:
            yield timer
    finally:
        print(f"Request took {timer.elapsed:.2f} seconds")
 
@app.get("/slow-operation")
async def slow_operation(timer: Timer = Depends(get_timer)):
    """Automatically timed."""
    # Simulate slow operation
    time.sleep(2)
    return {"message": "Done"}

from pathlib import Path
 
def get_temp_file():
    """Temporary file with automatic cleanup."""
    temp_file = Path("/tmp/temp_file.txt")
    temp_file.touch()
    try:
        yield temp_file
    finally:
        temp_file.unlink(missing_ok=True)
 
@app.post("/process-file/")
async def process_file(
    content: str,
    temp_file: Path = Depends(get_temp_file)
):
    """Process content with temporary file."""
    temp_file.write_text(content)
    # Process file...
    result = temp_file.read_text()
    # File automatically deleted after response
    return {"processed": result}

from typing import Optional
 
def pagination(page: int = 1, size: int = 50):
    """Pagination parameters with validation."""
    if page < 1:
        raise HTTPException(status_code=400, detail="Page must be >= 1")
    if size < 1 or size > 100:
        raise HTTPException(status_code=400, detail="Size must be 1-100")
    
    skip = (page - 1) * size
    return {"skip": skip, "limit": size}
 
@app.get("/items/")
async def list_items(
    paging: dict = Depends(pagination),
    db: Session = Depends(get_db)
):
    """List items with pagination."""
    items = db.query(Item).offset(paging["skip"]).limit(paging["limit"]).all()
    return items

from typing import Optional
 
class RateLimiter:
    """Rate limiting dependency."""
    
    def __init__(self, max_calls: int = 10, period: int = 60):
        self.max_calls = max_calls
        self.period = period
        self.calls = {}
    
    def __call__(self, request: Request):
        """Check rate limit for IP."""
        client_ip = request.client.host
        now = time.time()
        
        # Clean old entries
        self.calls = {
            ip: times 
            for ip, times in self.calls.items() 
            if times[-1] > now - self.period
        }
        
        # Check limit
        if client_ip in self.calls:
            if len(self.calls[client_ip]) >= self.max_calls:
                raise HTTPException(
                    status_code=429,
                    detail="Rate limit exceeded"
                )
            self.calls[client_ip].append(now)
        else:
            self.calls[client_ip] = [now]
        
        return True
 
# Create rate limiter instances
rate_limit_strict = RateLimiter(max_calls=5, period=60)
rate_limit_relaxed = RateLimiter(max_calls=100, period=60)
 
@app.get("/api/strict", dependencies=[Depends(rate_limit_strict)])
async def strict_endpoint():
    """Maximum 5 calls per minute."""
    return {"message": "Success"}
 
@app.get("/api/relaxed", dependencies=[Depends(rate_limit_relaxed)])
async def relaxed_endpoint():
    """Maximum 100 calls per minute."""
    return {"message": "Success"}

from functools import lru_cache
from pydantic_settings import BaseSettings
 
class Settings(BaseSettings):
    """Application settings."""
    app_name: str = "FastAPI App"
    database_url: str
    secret_key: str
    
    class Config:
        env_file = ".env"
 
@lru_cache()
def get_settings():
    """
    Cached settings - loaded only once.
    Perfect for configuration that doesn't change during runtime.
    """
    return Settings()
 
@app.get("/info")
async def info(settings: Settings = Depends(get_settings)):
    """Get app info - settings loaded from cache."""
    return {
        "app_name": settings.app_name,
        "database": settings.database_url
    }

# main.py
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

# test_main.py
from fastapi.testclient import TestClient
from main import app, get_db
 
# Mock database
def override_get_db():
    """Override database with test database."""
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()
 
# Override the dependency
app.dependency_overrides[get_db] = override_get_db
 
client = TestClient(app)
 
def test_get_user():
    response = client.get("/users/1")
    assert response.status_code == 200
    # Uses test database instead of production database!

from jose import JWTError, jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
 
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    """Extract and validate user from JWT token."""
    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])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise credentials_exception
    
    return user
 
async def get_current_active_user(
    current_user: User = Depends(get_current_user)
) -> User:
    """Ensure user is active."""
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user
 
async def get_current_admin_user(
    current_user: User = Depends(get_current_active_user)
) -> User:
    """Ensure user is admin."""
    if not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Not enough permissions")
    return current_user
 
# Use in routes
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user
 
@app.get("/admin/users")
async def read_all_users(
    admin_user: User = Depends(get_current_admin_user),
    db: Session = Depends(get_db)
):
    return db.query(User).all()

from typing import Optional
from pydantic import BaseModel
 
class PaginationParams(BaseModel):
    """Pagination parameters with validation."""
    skip: int = 0
    limit: int = 100
    
    def __init__(self, page: int = 1, page_size: int = 50):
        if page < 1:
            raise HTTPException(status_code=400, detail="Page must be >= 1")
        if page_size < 1 or page_size > 100:
            raise HTTPException(status_code=400, detail="Page size must be 1-100")
        
        skip = (page - 1) * page_size
        super().__init__(skip=skip, limit=page_size)
 
@app.get("/items/")
async def list_items(
    pagination: PaginationParams = Depends(),
    db: Session = Depends(get_db)
):
    items = (
        db.query(Item)
        .offset(pagination.skip)
        .limit(pagination.limit)
        .all()
    )
    total = db.query(Item).count()
    
    return {
        "items": items,
        "total": total,
        "page": (pagination.skip // pagination.limit) + 1,
        "page_size": pagination.limit
    }

from fastapi import Request
 
async def validate_content_type(request: Request):
    """Ensure request has correct content type."""
    content_type = request.headers.get("content-type")
    if content_type != "application/json":
        raise HTTPException(
            status_code=400,
            detail="Content-Type must be application/json"
        )
    return content_type
 
async def validate_request_size(request: Request):
    """Ensure request body is not too large."""
    content_length = request.headers.get("content-length")
    if content_length and int(content_length) > 1_000_000:  # 1MB
        raise HTTPException(
            status_code=413,
            detail="Request body too large"
        )
    return True
 
@app.post("/upload/", dependencies=[
    Depends(validate_content_type),
    Depends(validate_request_size)
])
async def upload_data(data: dict):
    """Upload with validation."""
    return {"received": data}

class UserService:
    """User business logic."""
    
    def __init__(self, db: Session = Depends(get_db)):
        self.db = db
    
    def get_user(self, user_id: int) -> User:
        user = self.db.query(User).filter(User.id == user_id).first()
        if not user:
            raise HTTPException(status_code=404, detail="User not found")
        return user
    
    def create_user(self, user_data: UserCreate) -> User:
        # Check if user exists
        existing = self.db.query(User).filter(
            User.email == user_data.email
        ).first()
        if existing:
            raise HTTPException(status_code=400, detail="Email already exists")
        
        # Create user
        user = User(**user_data.dict())
        self.db.add(user)
        self.db.commit()
        self.db.refresh(user)
        return user
    
    def update_user(self, user_id: int, user_data: UserUpdate) -> User:
        user = self.get_user(user_id)
        for field, value in user_data.dict(exclude_unset=True).items():
            setattr(user, field, value)
        self.db.commit()
        self.db.refresh(user)
        return user
 
# Use in routes
@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    user_service: UserService = Depends()
):
    return user_service.get_user(user_id)
 
@app.post("/users/")
async def create_user(
    user_data: UserCreate,
    user_service: UserService = Depends()
):
    return user_service.create_user(user_data)

class ConnectionPool:
    """Connection pool as a dependency."""
    
    def __init__(self):
        self.connections = []
        self.max_size = 10
    
    async def get_connection(self):
        """Get connection from pool."""
        if not self.connections:
            # Create new connection
            return await create_connection()
        return self.connections.pop()
    
    async def return_connection(self, conn):
        """Return connection to pool."""
        if len(self.connections) < self.max_size:
            self.connections.append(conn)
        else:
            await conn.close()
 
# Single instance for the app
pool = ConnectionPool()
 
async def get_connection():
    """Get connection from pool with automatic return."""
    conn = await pool.get_connection()
    try:
        yield conn
    finally:
        await pool.return_connection(conn)
 
@app.get("/data")
async def get_data(conn = Depends(get_connection)):
    return await conn.fetch("SELECT * FROM data")

from typing import Optional
 
def get_optional_user(
    token: Optional[str] = Header(None),
    db: Session = Depends(get_db)
) -> Optional[User]:
    """Get user if token provided, None otherwise."""
    if not token:
        return None
    
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        return db.query(User).filter(User.username == username).first()
    except JWTError:
        return None
 
@app.get("/items/")
async def list_items(
    user: Optional[User] = Depends(get_optional_user),
    db: Session = Depends(get_db)
):
    """List items - authenticated users see private items."""
    query = db.query(Item)
    
    if user:
        # Show all items for authenticated users
        query = query.filter(
            (Item.is_public == True) | (Item.owner_id == user.id)
        )
    else:
        # Show only public items
        query = query.filter(Item.is_public == True)
    
    return query.all()

from dataclasses import dataclass
 
@dataclass
class RequestContext:
    """Aggregated request context."""
    user: User
    db: Session
    settings: Settings
    
async def get_request_context(
    user: User = Depends(get_current_active_user),
    db: Session = Depends(get_db),
    settings: Settings = Depends(get_settings)
) -> RequestContext:
    """Combine multiple dependencies into one context."""
    return RequestContext(user=user, db=db, settings=settings)
 
@app.get("/profile")
async def get_profile(ctx: RequestContext = Depends(get_request_context)):
    """Use aggregated context."""
    # Access all dependencies through ctx
    user_data = ctx.db.query(UserProfile).filter(
        UserProfile.user_id == ctx.user.id
    ).first()
    
    return {
        "user": ctx.user,
        "profile": user_data,
        "app_name": ctx.settings.app_name
    }

# ✅ Good: Pure dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
 
# ❌ Bad: Dependency with side effects
def get_db():
    db = SessionLocal()
    log_access()  # Side effect!
    yield db
    db.close()

# ✅ Good: Clear types
def get_user(user_id: int, db: Session = Depends(get_db)) -> User:
    return db.query(User).filter(User.id == user_id).first()
 
# ❌ Bad: No type hints
def get_user(user_id, db = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

# ✅ Good: Clear names
get_db
get_current_user
verify_token
require_admin
 
# ❌ Bad: Vague names
dep1
check
validate
helper

# ❌ Bad: Dependency for simple parameter
def get_user_id(user_id: int = Path(...)):
    return user_id
 
@app.get("/users/{user_id}")
async def get_user(uid: int = Depends(get_user_id)):
    # Just use user_id directly!
    pass
 
# ✅ Good: Use parameter directly
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Simple and clear
    pass

# ✅ Good: Composed dependencies
def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    # Uses sub-dependencies
    pass
 
# ❌ Bad: Everything in one dependency
def get_current_user(
    authorization: str = Header(...),
) -> User:
    # Parsing token, connecting to DB, querying - all in one function
    # Hard to test and reuse
    pass

# conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
 
from main import app, get_db
from database import Base
 
# Test database
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
def override_get_db():
    """Test database dependency."""
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()
 
@pytest.fixture
def client():
    """Test client with overridden dependencies."""
    Base.metadata.create_all(bind=engine)
    app.dependency_overrides[get_db] = override_get_db
    
    with TestClient(app) as c:
        yield c
    
    Base.metadata.drop_all(bind=engine)
    app.dependency_overrides.clear()
 
# test_users.py
def test_create_user(client):
    response = client.post("/users/", json={
        "username": "test",
        "email": "[email protected]"
    })
    assert response.status_code == 201
    # Uses test database!