第二章练习答案

练习 1：数据库连接依赖

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

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

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

@app.get("/users/")
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = db.query(User).offset(skip).limit(limit).all()
    return users

练习 2：分页依赖

class Pagination:
    def __init__(self, page: int = 1, page_size: int = 10):
        self.page = page
        self.page_size = page_size
        self.skip = (page - 1) * page_size
        self.limit = page_size

def get_pagination(page: int = 1, page_size: int = 10):
    return Pagination(page, page_size)

@app.get("/items/")
def list_items(pagination: Pagination = Depends(get_pagination)):
    items = get_items(skip=pagination.skip, limit=pagination.limit)
    return {
        "page": pagination.page,
        "page_size": pagination.page_size,
        "items": items
    }

练习 3：多级认证

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    if not is_valid_token(token):
        raise HTTPException(status_code=401, detail="Invalid token")
    return decode_token(token)

def get_current_user(token_data: dict = Depends(verify_token)):
    user = get_user_by_id(token_data["user_id"])
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

def require_admin(user: User = Depends(get_current_user)):
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin required")
    return user

@app.delete("/users/{user_id}")
def delete_user(user_id: int, admin: User = Depends(require_admin)):
    delete_user_by_id(user_id)
    return {"message": "User deleted"}

返回第二章概览