FastAPI实战项目案例分享从API设计到系统优化涵盖认证授权数据库集成缓存策略异步处理等关键技术点助你掌握现代Web开发精髓

威震华夏关云长

引言

FastAPI是近年来备受瞩目的Python Web框架，它凭借出色的性能、直观的API设计和现代化的开发体验，迅速成为构建高性能API的首选框架之一。本文将通过一个完整的实战项目，深入探讨FastAPI的各项关键技术，帮助开发者从零开始构建一个生产级别的Web应用。

FastAPI的核心优势在于：

• 基于Starlette和Pydantic，提供高性能的异步处理能力
• 自动生成交互式API文档（Swagger UI）
• 内置数据验证和序列化功能
• 支持依赖注入系统，便于测试和模块化开发
• 类型提示支持，提供更好的IDE体验和代码质量

在本文中，我们将通过构建一个完整的博客系统来展示FastAPI的强大功能，涵盖从基础API设计到高级系统优化的全过程。

项目结构设计

良好的项目结构是成功的一半。对于FastAPI项目，推荐采用以下结构：

2. blog_system/
3. ├── app/
4. │   ├── __init__.py
5. │   ├── main.py           # FastAPI应用入口
6. │   ├── core/             # 核心配置
7. │   │   ├── __init__.py
8. │   │   ├── config.py     # 配置管理
9. │   │   └── security.py   # 安全相关功能
10. │   ├── api/              # API路由
11. │   │   ├── __init__.py
12. │   │   ├── v1/
13. │   │   │   ├── __init__.py
14. │   │   │   ├── endpoints/
15. │   │   │   │   ├── __init__.py
16. │   │   │   │   ├── posts.py
17. │   │   │   │   ├── users.py
18. │   │   │   │   └── comments.py
19. │   │   │   └── api.py    # API路由注册
20. │   ├── models/           # 数据模型
21. │   │   ├── __init__.py
22. │   │   ├── user.py
23. │   │   ├── post.py
24. │   │   └── comment.py
25. │   ├── schemas/          # Pydantic模型
26. │   │   ├── __init__.py
27. │   │   ├── user.py
28. │   │   ├── post.py
29. │   │   └── comment.py
30. │   ├── crud/             # 数据库操作
31. │   │   ├── __init__.py
32. │   │   ├── base.py
33. │   │   ├── user.py
34. │   │   ├── post.py
35. │   │   └── comment.py
36. │   ├── db/               # 数据库配置
37. │   │   ├── __init__.py
38. │   │   ├── session.py    # 数据库会话
39. │   │   └── base_class.py # 基础模型类
40. │   └── utils/            # 工具函数
41. │       ├── __init__.py
42. │       └── helpers.py
43. ├── tests/                # 测试目录
44. │   ├── __init__.py
45. │   ├── conftest.py
46. │   ├── test_api.py
47. │   └── test_crud.py
48. ├── alembic/              # 数据库迁移
49. ├── requirements.txt      # 项目依赖
50. ├── .env                  # 环境变量
51. └── Dockerfile            # Docker配置

复制代码

这种结构遵循了领域驱动设计(DDD)的原则，将不同功能的代码分门别类地组织在一起，便于维护和扩展。

API设计与实现

RESTful API设计原则

在FastAPI中设计API时，我们遵循RESTful原则：

1. 使用HTTP动词表示操作类型（GET、POST、PUT、DELETE等）
2. 使用名词表示资源（如/users、/posts）
3. 使用嵌套资源表示关系（如/posts/{post_id}/comments）
4. 使用HTTP状态码表示操作结果
5. 支持内容协商（JSON、XML等）

基本路由实现

让我们从实现博客系统的基本路由开始：

2. # app/main.py
3. from fastapi import FastAPI
4. from app.api.v1.api import api_router
5. from app.core.config import settings
6. from fastapi.middleware.cors import CORSMiddleware
8. app = FastAPI(
9.     title=settings.PROJECT_NAME,
10.     openapi_url=f"{settings.API_V1_STR}/openapi.json"
11. )
13. # 设置CORS
14. if settings.BACKEND_CORS_ORIGINS:
15.     app.add_middleware(
16.         CORSMiddleware,
17.         allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS],
18.         allow_credentials=True,
19.         allow_methods=["*"],
20.         allow_headers=["*"],
21.     )
23. app.include_router(api_router, prefix=settings.API_V1_STR)
25. @app.get("/")
26. async def root():
27.     return {"message": "Welcome to Blog System API"}

复制代码

2. # app/api/v1/api.py
3. from fastapi import APIRouter
4. from app.api.v1.endpoints import posts, users, comments
6. api_router = APIRouter()
7. api_router.include_router(users.router, prefix="/users", tags=["users"])
8. api_router.include_router(posts.router, prefix="/posts", tags=["posts"])
9. api_router.include_router(comments.router, prefix="/comments", tags=["comments"])

复制代码

2. # app/api/v1/endpoints/posts.py
3. from typing import List, Optional
4. from fastapi import APIRouter, Depends, HTTPException, Query
5. from sqlalchemy.orm import Session
6. from app import crud, models, schemas
7. from app.api import deps
9. router = APIRouter()
11. @router.get("/", response_model=List[schemas.Post])
12. def read_posts(
13.     db: Session = Depends(deps.get_db),
14.     skip: int = 0,
15.     limit: int = 100,
16.     search: Optional[str] = Query(None, min_length=3, description="Search in post title and content")
17. ):
18.     """
19.     Retrieve posts.
20.     """
21.     if search:
22.         posts = crud.post.search(db, search=search, skip=skip, limit=limit)
23.     else:
24.         posts = crud.post.get_multi(db, skip=skip, limit=limit)
25.     return posts
27. @router.post("/", response_model=schemas.Post)
28. def create_post(
29.     *,
30.     db: Session = Depends(deps.get_db),
31.     post_in: schemas.PostCreate,
32.     current_user: models.User = Depends(deps.get_current_active_user),
33. ):
34.     """
35.     Create new post.
36.     """
37.     post = crud.post.create_with_owner(db=db, obj_in=post_in, owner_id=current_user.id)
38.     return post
40. @router.get("/{post_id}", response_model=schemas.Post)
41. def read_post(
42.     *,
43.     db: Session = Depends(deps.get_db),
44.     post_id: int,
45. ):
46.     """
47.     Get post by ID.
48.     """
49.     post = crud.post.get(db=db, id=post_id)
50.     if not post:
51.         raise HTTPException(status_code=404, detail="Post not found")
52.     return post
54. @router.put("/{post_id}", response_model=schemas.Post)
55. def update_post(
56.     *,
57.     db: Session = Depends(deps.get_db),
58.     post_id: int,
59.     post_in: schemas.PostUpdate,
60.     current_user: models.User = Depends(deps.get_current_active_user),
61. ):
62.     """
63.     Update a post.
64.     """
65.     post = crud.post.get(db=db, id=post_id)
66.     if not post:
67.         raise HTTPException(status_code=404, detail="Post not found")
68.     if post.owner_id != current_user.id:
69.         raise HTTPException(status_code=403, detail="Not enough permissions")
70.     post = crud.post.update(db=db, db_obj=post, obj_in=post_in)
71.     return post
73. @router.delete("/{post_id}", response_model=schemas.Post)
74. def delete_post(
75.     *,
76.     db: Session = Depends(deps.get_db),
77.     post_id: int,
78.     current_user: models.User = Depends(deps.get_current_active_user),
79. ):
80.     """
81.     Delete a post.
82.     """
83.     post = crud.post.get(db=db, id=post_id)
84.     if not post:
85.         raise HTTPException(status_code=404, detail="Post not found")
86.     if post.owner_id != current_user.id:
87.         raise HTTPException(status_code=403, detail="Not enough permissions")
88.     post = crud.post.remove(db=db, id=post_id)
89.     return post

复制代码

数据模型与Pydantic Schema

在FastAPI中，我们使用SQLAlchemy定义数据库模型，使用Pydantic定义请求和响应的数据结构：

2. # app/models/post.py
3. from sqlalchemy import Boolean, Column, Integer, String, Text, ForeignKey, DateTime
4. from sqlalchemy.orm import relationship
5. from sqlalchemy.sql import func
6. from app.db.base_class import Base
8. class Post(Base):
9.     id = Column(Integer, primary_key=True, index=True)
10.     title = Column(String(256), nullable=False, index=True)
11.     content = Column(Text, nullable=False)
12.     published = Column(Boolean, default=False)
13.     created_at = Column(DateTime(timezone=True), server_default=func.now())
14.     updated_at = Column(DateTime(timezone=True), onupdate=func.now())
15.     owner_id = Column(Integer, ForeignKey("user.id"), nullable=False)
16.    
17.     owner = relationship("User", back_populates="posts")
18.     comments = relationship("Comment", back_populates="post", cascade="all, delete-orphan")

复制代码

2. # app/schemas/post.py
3. from typing import Optional, List
4. from datetime import datetime
5. from pydantic import BaseModel
7. # Base schema with common attributes
8. class PostBase(BaseModel):
9.     title: str
10.     content: str
11.     published: bool = False
13. # Schema for creating a new post
14. class PostCreate(PostBase):
15.     pass
17. # Schema for updating a post
18. class PostUpdate(PostBase):
19.     title: Optional[str] = None
20.     content: Optional[str] = None
21.     published: Optional[bool] = None
23. # Schema for returning post data
24. class Post(PostBase):
25.     id: int
26.     created_at: datetime
27.     updated_at: Optional[datetime] = None
28.     owner_id: int
29.    
30.     class Config:
31.         orm_mode = True

复制代码

认证与授权

JWT认证实现

在FastAPI中实现JWT认证非常简单，我们可以使用OAuth2密码流和JWT令牌：

2. # app/core/security.py
3. from datetime import datetime, timedelta
4. from typing import Any, Union, Optional
5. from jose import jwt
6. from passlib.context import CryptContext
7. from app.core.config import settings
9. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
11. def create_access_token(
12.     subject: Union[str, Any], expires_delta: Optional[timedelta] = None
13. ) -> str:
14.     """
15.     创建JWT访问令牌
16.     """
17.     if expires_delta:
18.         expire = datetime.utcnow() + expires_delta
19.     else:
20.         expire = datetime.utcnow() + timedelta(
21.             minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
22.         )
23.     to_encode = {"exp": expire, "sub": str(subject)}
24.     encoded_jwt = jwt.encode(
25.         to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
26.     )
27.     return encoded_jwt
29. def verify_password(plain_password: str, hashed_password: str) -> bool:
30.     """
31.     验证密码
32.     """
33.     return pwd_context.verify(plain_password, hashed_password)
35. def get_password_hash(password: str) -> str:
36.     """
37.     获取密码哈希值
38.     """
39.     return pwd_context.hash(password)

复制代码

2. # app/api/v1/endpoints/users.py
3. from datetime import timedelta
4. from typing import Any
5. from fastapi import APIRouter, Depends, HTTPException
6. from fastapi.security import OAuth2PasswordRequestForm
7. from sqlalchemy.orm import Session
8. from app import crud, models, schemas
9. from app.api import deps
10. from app.core import security
11. from app.core.config import settings
13. router = APIRouter()
15. @router.post("/login/access-token", response_model=schemas.Token)
16. def login_access_token(
17.     db: Session = Depends(deps.get_db),
18.     form_data: OAuth2PasswordRequestForm = Depends()
19. ) -> Any:
20.     """
21.     OAuth2 compatible token login, get an access token for future requests
22.     """
23.     user = crud.user.authenticate(
24.         db, email=form_data.username, password=form_data.password
25.     )
26.     if not user:
27.         raise HTTPException(status_code=400, detail="Incorrect email or password")
28.     elif not crud.user.is_active(user):
29.         raise HTTPException(status_code=400, detail="Inactive user")
30.     access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
31.     return {
32.         "access_token": security.create_access_token(
33.             user.id, expires_delta=access_token_expires
34.         ),
35.         "token_type": "bearer",
36.     }

复制代码

权限控制

FastAPI的依赖注入系统使得实现权限控制变得非常简单：

2. # app/api/deps.py
3. from fastapi import Depends, HTTPException, status
4. from fastapi.security import OAuth2PasswordBearer
5. from jose import jwt
6. from pydantic import ValidationError
7. from sqlalchemy.orm import Session
8. from app import crud, models, schemas
9. from app.core import security
10. from app.core.config import settings
11. from app.db.session import SessionLocal
13. oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/login/access-token")
15. def get_db() -> Session:
16.     """
17.     获取数据库会话
18.     """
19.     db = SessionLocal()
20.     try:
21.         yield db
22.     finally:
23.         db.close()
25. def get_current_user(
26.     db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
27. ) -> models.User:
28.     """
29.     获取当前用户
30.     """
31.     try:
32.         payload = jwt.decode(
33.             token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
34.         )
35.         token_data = schemas.TokenPayload(**payload)
36.     except (jwt.JWTError, ValidationError):
37.         raise HTTPException(
38.             status_code=status.HTTP_403_FORBIDDEN,
39.             detail="Could not validate credentials",
40.         )
41.     user = crud.user.get(db, id=token_data.sub)
42.     if not user:
43.         raise HTTPException(status_code=404, detail="User not found")
44.     return user
46. def get_current_active_user(
47.     current_user: models.User = Depends(get_current_user),
48. ) -> models.User:
49.     """
50.     获取当前活跃用户
51.     """
52.     if not crud.user.is_active(current_user):
53.         raise HTTPException(status_code=400, detail="Inactive user")
54.     return current_user
56. def get_current_active_superuser(
57.     current_user: models.User = Depends(get_current_user),
58. ) -> models.User:
59.     """
60.     获取当前活跃超级用户
61.     """
62.     if not crud.user.is_superuser(current_user):
63.         raise HTTPException(
64.             status_code=400, detail="The user doesn't have enough privileges"
65.         )
66.     return current_user

复制代码

使用这些依赖项，我们可以轻松地保护API端点：

2. # app/api/v1/endpoints/users.py
3. @router.get("/me", response_model=schemas.User)
4. def read_user_me(
5.     current_user: models.User = Depends(deps.get_current_active_user),
6. ):
7.     """
8.     获取当前用户信息
9.     """
10.     return current_user
12. @router.post("/", response_model=schemas.User)
13. def create_user(
14.     *,
15.     db: Session = Depends(deps.get_db),
16.     user_in: schemas.UserCreate,
17.     current_user: models.User = Depends(deps.get_current_active_superuser),
18. ):
19.     """
20.     创建新用户（仅限超级用户）
21.     """
22.     user = crud.user.get_by_email(db, email=user_in.email)
23.     if user:
24.         raise HTTPException(
25.             status_code=400,
26.             detail="The user with this email already exists in the system.",
27.         )
28.     user = crud.user.create(db, obj_in=user_in)
29.     return user

复制代码

数据库集成

SQLAlchemy配置

FastAPI与SQLAlchemy的集成非常顺畅，我们可以使用异步SQLAlchemy来提高性能：

2. # app/db/session.py
3. from sqlalchemy import create_engine
4. from sqlalchemy.ext.declarative import declarative_base
5. from sqlalchemy.orm import sessionmaker
6. from app.core.config import settings
8. # 创建同步SQLAlchemy引擎
9. engine = create_engine(settings.SQLALCHEMY_DATABASE_URI, pool_pre_ping=True)
10. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
12. # 创建异步SQLAlchemy引擎
13. from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
14. async_engine = create_async_engine(settings.SQLALCHEMY_ASYNC_DATABASE_URI)
15. AsyncSessionLocal = sessionmaker(
16.     async_engine, class_=AsyncSession, expire_on_commit=False
17. )
19. # 基础模型类
20. Base = declarative_base()
22. # 获取数据库会话的依赖项
23. def get_db() -> Session:
24.     db = SessionLocal()
25.     try:
26.         yield db
27.     finally:
28.         db.close()
30. # 获取异步数据库会话的依赖项
31. async def get_async_db() -> AsyncSession:
32.     async with AsyncSessionLocal() as session:
33.         try:
34.             yield session
35.         finally:
36.             await session.close()

复制代码

CRUD操作

使用SQLAlchemy ORM进行CRUD操作：

2. # app/crud/base.py
3. from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
4. from fastapi.encoders import jsonable_encoder
5. from pydantic import BaseModel
6. from sqlalchemy.orm import Session
7. from app.db.base_class import Base
9. ModelType = TypeVar("ModelType", bound=Base)
10. CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
11. UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
13. class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
14.     def __init__(self, model: Type[ModelType]):
15.         """
16.         CRUD object with default methods to Create, Read, Update, Delete (CRUD).
17.         """
18.         self.model = model
20.     def get(self, db: Session, id: Any) -> Optional[ModelType]:
21.         return db.query(self.model).filter(self.model.id == id).first()
23.     def get_multi(
24.         self, db: Session, *, skip: int = 0, limit: int = 100
25.     ) -> List[ModelType]:
26.         return db.query(self.model).offset(skip).limit(limit).all()
28.     def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
29.         obj_in_data = jsonable_encoder(obj_in)
30.         db_obj = self.model(**obj_in_data)  # type: ignore
31.         db.add(db_obj)
32.         db.commit()
33.         db.refresh(db_obj)
34.         return db_obj
36.     def update(
37.         self,
38.         db: Session,
39.         *,
40.         db_obj: ModelType,
41.         obj_in: Union[UpdateSchemaType, Dict[str, Any]]
42.     ) -> ModelType:
43.         obj_data = jsonable_encoder(db_obj)
44.         if isinstance(obj_in, dict):
45.             update_data = obj_in
46.         else:
47.             update_data = obj_in.dict(exclude_unset=True)
48.         for field in obj_data:
49.             if field in update_data:
50.                 setattr(db_obj, field, update_data[field])
51.         db.add(db_obj)
52.         db.commit()
53.         db.refresh(db_obj)
54.         return db_obj
56.     def remove(self, db: Session, *, id: int) -> ModelType:
57.         obj = db.query(self.model).get(id)
58.         db.delete(obj)
59.         db.commit()
60.         return obj

复制代码

2. # app/crud/post.py
3. from typing import List, Optional
4. from sqlalchemy.orm import Session
5. from sqlalchemy import or_
6. from app.crud.base import CRUDBase
7. from app.models.post import Post
8. from app.schemas.post import PostCreate, PostUpdate
10. class CRUDPost(CRUDBase[Post, PostCreate, PostUpdate]):
11.     def create_with_owner(
12.         self, db: Session, *, obj_in: PostCreate, owner_id: int
13.     ) -> Post:
14.         obj_in_data = obj_in.dict()
15.         db_obj = Post(**obj_in_data, owner_id=owner_id)
16.         db.add(db_obj)
17.         db.commit()
18.         db.refresh(db_obj)
19.         return db_obj
20.    
21.     def get_multi_by_owner(
22.         self, db: Session, *, owner_id: int, skip: int = 0, limit: int = 100
23.     ) -> List[Post]:
24.         return (
25.             db.query(self.model)
26.             .filter(Post.owner_id == owner_id)
27.             .offset(skip)
28.             .limit(limit)
29.             .all()
30.         )
31.    
32.     def search(
33.         self, db: Session, *, search: str, skip: int = 0, limit: int = 100
34.     ) -> List[Post]:
35.         return (
36.             db.query(self.model)
37.             .filter(
38.                 or_(
39.                     Post.title.ilike(f"%{search}%"),
40.                     Post.content.ilike(f"%{search}%")
41.                 )
42.             )
43.             .offset(skip)
44.             .limit(limit)
45.             .all()
46.         )
48. post = CRUDPost(Post)

复制代码

异步数据库操作

FastAPI支持异步操作，我们可以使用SQLAlchemy的异步API来提高性能：

2. # app/crud/async_post.py
3. from typing import List, Optional
4. from sqlalchemy.ext.asyncio import AsyncSession
5. from sqlalchemy import select, or_
6. from app.models.post import Post
7. from app.schemas.post import PostCreate, PostUpdate
9. async def async_get_post(db: AsyncSession, post_id: int) -> Optional[Post]:
10.     result = await db.execute(select(Post).where(Post.id == post_id))
11.     return result.scalar_one_or_none()
13. async def async_get_posts(
14.     db: AsyncSession, skip: int = 0, limit: int = 100
15. ) -> List[Post]:
16.     result = await db.execute(select(Post).offset(skip).limit(limit))
17.     return result.scalars().all()
19. async def async_create_post(
20.     db: AsyncSession, *, post_in: PostCreate, owner_id: int
21. ) -> Post:
22.     db_post = Post(**post_in.dict(), owner_id=owner_id)
23.     db.add(db_post)
24.     await db.commit()
25.     await db.refresh(db_post)
26.     return db_post
28. async def async_search_posts(
29.     db: AsyncSession, *, search: str, skip: int = 0, limit: int = 100
30. ) -> List[Post]:
31.     result = await db.execute(
32.         select(Post)
33.         .where(
34.             or_(
35.                 Post.title.ilike(f"%{search}%"),
36.                 Post.content.ilike(f"%{search}%")
37.             )
38.         )
39.         .offset(skip)
40.         .limit(limit)
41.     )
42.     return result.scalars().all()

复制代码

缓存策略

Redis缓存集成

缓存是提高API性能的重要手段，我们可以使用Redis来实现缓存：

2. # app/core/redis.py
3. import redis
4. import json
5. from app.core.config import settings
7. # 创建Redis连接
8. redis_client = redis.Redis(
9.     host=settings.REDIS_HOST,
10.     port=settings.REDIS_PORT,
11.     db=settings.REDIS_DB,
12.     password=settings.REDIS_PASSWORD,
13.     decode_responses=True
14. )
16. def get_from_cache(key: str):
17.     """
18.     从缓存中获取数据
19.     """
20.     try:
21.         data = redis_client.get(key)
22.         if data:
23.             return json.loads(data)
24.         return None
25.     except Exception as e:
26.         print(f"Redis GET error: {e}")
27.         return None
29. def set_to_cache(key: str, value: dict, expire: int = 3600):
30.     """
31.     设置缓存数据
32.     """
33.     try:
34.         redis_client.setex(key, expire, json.dumps(value))
35.     except Exception as e:
36.         print(f"Redis SET error: {e}")
38. def delete_from_cache(key: str):
39.     """
40.     删除缓存数据
41.     """
42.     try:
43.         redis_client.delete(key)
44.     except Exception as e:
45.         print(f"Redis DELETE error: {e}")

复制代码

缓存装饰器

我们可以创建一个缓存装饰器，方便在API端点中使用缓存：

2. # app/utils/cache.py
3. import json
4. from functools import wraps
5. from fastapi import Request, Response
6. from app.core.redis import get_from_cache, set_to_cache, delete_from_cache
8. def cache_response(expire: int = 3600):
9.     """
10.     缓存响应装饰器
11.     """
12.     def decorator(func):
13.         @wraps(func)
14.         async def wrapper(*args, **kwargs):
15.             # 尝试从kwargs中获取request对象
16.             request = None
17.             for arg in args:
18.                 if isinstance(arg, Request):
19.                     request = arg
20.                     break
21.             
22.             if request:
23.                 # 生成缓存键
24.                 cache_key = f"{request.url.path}:{json.dumps(request.query_params)}"
25.                
26.                 # 尝试从缓存获取数据
27.                 cached_data = get_from_cache(cache_key)
28.                 if cached_data:
29.                     return cached_data
30.                
31.                 # 执行原始函数
32.                 response = await func(*args, **kwargs)
33.                
34.                 # 将响应存入缓存
35.                 if isinstance(response, Response):
36.                     set_to_cache(cache_key, response.body.decode(), expire)
37.                 else:
38.                     set_to_cache(cache_key, response, expire)
39.                
40.                 return response
41.             else:
42.                 # 如果没有request对象，直接执行原始函数
43.                 return await func(*args, **kwargs)
44.         
45.         return wrapper
46.     return decorator
48. def invalidate_cache(pattern: str):
49.     """
50.     缓存失效装饰器
51.     """
52.     def decorator(func):
53.         @wraps(func)
54.         async def wrapper(*args, **kwargs):
55.             # 执行原始函数
56.             result = await func(*args, **kwargs)
57.             
58.             # 使匹配模式的缓存失效
59.             try:
60.                 from app.core.redis import redis_client
61.                 for key in redis_client.scan_iter(match=pattern):
62.                     redis_client.delete(key)
63.             except Exception as e:
64.                 print(f"Cache invalidation error: {e}")
65.             
66.             return result
67.         
68.         return wrapper
69.     return decorator

复制代码

在API中使用缓存

现在我们可以在API端点中使用缓存：

2. # app/api/v1/endpoints/posts.py
3. from fastapi import APIRouter, Depends, HTTPException, Query, Request
4. from sqlalchemy.orm import Session
5. from app import crud, models, schemas
6. from app.api import deps
7. from app.utils.cache import cache_response, invalidate_cache
9. router = APIRouter()
11. @router.get("/", response_model=List[schemas.Post])
12. @cache_response(expire=1800)  # 缓存30分钟
13. def read_posts(
14.     request: Request,
15.     db: Session = Depends(deps.get_db),
16.     skip: int = 0,
17.     limit: int = 100,
18.     search: str = None,
19. ):
20.     """
21.     检索文章列表，支持缓存
22.     """
23.     if search:
24.         posts = crud.post.search(db, search=search, skip=skip, limit=limit)
25.     else:
26.         posts = crud.post.get_multi(db, skip=skip, limit=limit)
27.     return posts
29. @router.post("/", response_model=schemas.Post)
30. @invalidate_cache(pattern="/api/v1/posts*")  # 使文章列表缓存失效
31. def create_post(
32.     *,
33.     db: Session = Depends(deps.get_db),
34.     post_in: schemas.PostCreate,
35.     current_user: models.User = Depends(deps.get_current_active_user),
36. ):
37.     """
38.     创建新文章，并使相关缓存失效
39.     """
40.     post = crud.post.create_with_owner(db=db, obj_in=post_in, owner_id=current_user.id)
41.     return post

复制代码

异步处理

异步任务处理

FastAPI原生支持异步操作，我们可以使用它来处理长时间运行的任务：

2. # app/utils/tasks.py
3. import asyncio
4. from typing import Dict, Any
5. from fastapi import BackgroundTasks
6. from app.core.redis import redis_client
7. import json
9. async def process_data(data: Dict[str, Any]):
10.     """
11.     模拟一个长时间运行的数据处理任务
12.     """
13.     # 模拟耗时操作
14.     await asyncio.sleep(10)
15.    
16.     # 处理数据
17.     processed_data = {k: v.upper() if isinstance(v, str) else v for k, v in data.items()}
18.    
19.     # 将结果存储到Redis
20.     task_id = data.get("task_id")
21.     if task_id:
22.         redis_client.setex(f"task_result:{task_id}", 3600, json.dumps(processed_data))
23.    
24.     return processed_data
26. def run_task_in_background(background_tasks: BackgroundTasks, data: Dict[str, Any]):
27.     """
28.     在后台运行任务
29.     """
30.     background_tasks.add_task(process_data, data)

复制代码

2. # app/api/v1/endpoints/tasks.py
3. from fastapi import APIRouter, BackgroundTasks, HTTPException
4. from app.utils.tasks import run_task_in_background
5. from app.core.redis import redis_client
6. import json
7. import uuid
9. router = APIRouter()
11. @router.post("/process")
12. async def create_processing_task(background_tasks: BackgroundTasks):
13.     """
14.     创建一个数据处理任务
15.     """
16.     task_id = str(uuid.uuid4())
17.     data = {"task_id": task_id, "name": "sample data", "value": 42}
18.    
19.     # 在后台运行任务
20.     run_task_in_background(background_tasks, data)
21.    
22.     return {"task_id": task_id, "message": "Task started"}
24. @router.get("/status/{task_id}")
25. async def get_task_status(task_id: str):
26.     """
27.     获取任务状态和结果
28.     """
29.     result = redis_client.get(f"task_result:{task_id}")
30.     if result:
31.         return {"task_id": task_id, "status": "completed", "result": json.loads(result)}
32.     else:
33.         return {"task_id": task_id, "status": "processing"}

复制代码

使用Celery进行分布式任务处理

对于更复杂的异步任务处理，我们可以集成Celery：

2. # app/core/celery_app.py
3. from celery import Celery
4. from app.core.config import settings
6. # 创建Celery实例
7. celery_app = Celery(
8.     "blog_system",
9.     broker=settings.CELERY_BROKER_URL,
10.     backend=settings.CELERY_RESULT_BACKEND,
11.     include=["app.tasks"]
12. )
14. # Celery配置
15. celery_app.conf.update(
16.     task_serializer="json",
17.     result_serializer="json",
18.     accept_content=["json"],
19.     timezone="UTC",
20.     enable_utc=True,
21.     task_track_started=True,
22.     task_time_limit=30 * 60,  # 30分钟超时
23.     task_soft_time_limit=25 * 60,  # 25分钟软超时
24.     worker_prefetch_multiplier=1,  # 每个worker一次只取一个任务
25. )

复制代码

2. # app/tasks.py
3. from app.core.celery_app import celery_app
4. from app.core.redis import redis_client
5. import json
6. import time
8. @celery_app.task(bind=True)
9. def process_data_task(self, data):
10.     """
11.     使用Celery处理数据的任务
12.     """
13.     try:
14.         # 更新任务状态
15.         self.update_state(
16.             state="PROGRESS",
17.             meta={"current": 1, "total": 3, "status": "Starting data processing"}
18.         )
19.         
20.         # 模拟数据处理步骤1
21.         time.sleep(5)
22.         processed_data = {k: v.upper() if isinstance(v, str) else v for k, v in data.items()}
23.         
24.         # 更新任务状态
25.         self.update_state(
26.             state="PROGRESS",
27.             meta={"current": 2, "total": 3, "status": "Processing data"}
28.         )
29.         
30.         # 模拟数据处理步骤2
31.         time.sleep(5)
32.         processed_data["processed_at"] = time.time()
33.         
34.         # 更新任务状态
35.         self.update_state(
36.             state="PROGRESS",
37.             meta={"current": 3, "total": 3, "status": "Saving results"}
38.         )
39.         
40.         # 模拟数据处理步骤3
41.         time.sleep(5)
42.         task_id = data.get("task_id")
43.         if task_id:
44.             redis_client.setex(f"celery_task_result:{task_id}", 3600, json.dumps(processed_data))
45.         
46.         # 返回结果
47.         return {"status": "completed", "result": processed_data}
48.    
49.     except Exception as e:
50.         # 更新任务状态为失败
51.         self.update_state(
52.             state="FAILURE",
53.             meta={"current": 3, "total": 3, "status": f"Error: {str(e)}"}
54.         )
55.         raise e

复制代码

2. # app/api/v1/endpoints/celery_tasks.py
3. from fastapi import APIRouter, HTTPException
4. from app.tasks import process_data_task
5. from app.core.redis import redis_client
6. import json
7. import uuid
9. router = APIRouter()
11. @router.post("/process")
12. async def create_celery_task():
13.     """
14.     创建一个Celery数据处理任务
15.     """
16.     task_id = str(uuid.uuid4())
17.     data = {"task_id": task_id, "name": "sample data", "value": 42}
18.    
19.     # 启动Celery任务
20.     task = process_data_task.delay(data)
21.    
22.     return {"task_id": task_id, "celery_task_id": task.id, "message": "Task started"}
24. @router.get("/status/{task_id}")
25. async def get_celery_task_status(task_id: str):
26.     """
27.     获取Celery任务状态和结果
28.     """
29.     # 从Redis获取结果
30.     result = redis_client.get(f"celery_task_result:{task_id}")
31.     if result:
32.         return {"task_id": task_id, "status": "completed", "result": json.loads(result)}
33.    
34.     # 如果Redis中没有结果，返回任务处理中
35.     return {"task_id": task_id, "status": "processing"}

复制代码

性能优化

数据库优化

数据库查询优化是提高API性能的关键：

2. # app/crud/optimized_post.py
3. from typing import List, Optional
4. from sqlalchemy.orm import Session, joinedload
5. from sqlalchemy import func, or_
6. from app.models.post import Post
7. from app.models.user import User
8. from app.schemas.post import PostCreate, PostUpdate
10. def get_post_with_author(db: Session, post_id: int) -> Optional[Post]:
11.     """
12.     获取文章及其作者信息（使用joinedload优化查询）
13.     """
14.     return (
15.         db.query(Post)
16.         .options(joinedload(Post.owner))
17.         .filter(Post.id == post_id)
18.         .first()
19.     )
21. def get_posts_with_pagination(
22.     db: Session, skip: int = 0, limit: int = 100
23. ) -> tuple[List[Post], int]:
24.     """
25.     获取文章列表及总数（用于分页）
26.     """
27.     # 使用一个查询获取总数
28.     total = db.query(func.count(Post.id)).scalar()
29.    
30.     # 使用另一个查询获取分页数据
31.     posts = (
32.         db.query(Post)
33.         .options(joinedload(Post.owner))
34.         .offset(skip)
35.         .limit(limit)
36.         .all()
37.     )
38.    
39.     return posts, total
41. def get_popular_posts(db: Session, limit: int = 10) -> List[Post]:
42.     """
43.     获取热门文章（基于评论数量）
44.     """
45.     return (
46.         db.query(Post)
47.         .join(Post.comments)
48.         .group_by(Post.id)
49.         .order_by(func.count(Post.comments).desc())
50.         .limit(limit)
51.         .all()
52.     )
54. def search_posts_optimized(
55.     db: Session, *, search: str, skip: int = 0, limit: int = 100
56. ) -> List[Post]:
57.     """
58.     优化的文章搜索（使用全文索引）
59.     """
60.     return (
61.         db.query(Post)
62.         .filter(
63.             or_(
64.                 Post.title.ilike(f"%{search}%"),
65.                 Post.content.ilike(f"%{search}%")
66.             )
67.         )
68.         .options(joinedload(Post.owner))
69.         .offset(skip)
70.         .limit(limit)
71.         .all()
72.     )

复制代码

响应压缩

FastAPI支持响应压缩，可以显著减少传输数据量：

2. # app/main.py
3. from fastapi import FastAPI
4. from fastapi.middleware.gzip import GZipMiddleware
5. from app.api.v1.api import api_router
6. from app.core.config import settings
8. app = FastAPI(
9.     title=settings.PROJECT_NAME,
10.     openapi_url=f"{settings.API_V1_STR}/openapi.json"
11. )
13. # 添加GZip中间件
14. app.add_middleware(GZipMiddleware, minimum_size=1000)
16. app.include_router(api_router, prefix=settings.API_V1_STR)

复制代码

异步端点

使用异步端点可以提高并发性能：

2. # app/api/v1/endpoints/async_posts.py
3. from typing import List, Optional
4. from fastapi import APIRouter, Depends, HTTPException, Query
5. from sqlalchemy.ext.asyncio import AsyncSession
6. from sqlalchemy import select, or_, func
7. from app.models.post import Post
8. from app.models.user import User
9. from app.schemas.post import Post, PostCreate
10. from app.api.deps import get_async_db
11. from app.crud.async_post import async_create_post
13. router = APIRouter()
15. @router.get("/", response_model=List[Post])
16. async def read_posts(
17.     db: AsyncSession = Depends(get_async_db),
18.     skip: int = 0,
19.     limit: int = 100,
20.     search: Optional[str] = Query(None, min_length=3),
21. ):
22.     """
23.     异步获取文章列表
24.     """
25.     if search:
26.         query = (
27.             select(Post)
28.             .where(
29.                 or_(
30.                     Post.title.ilike(f"%{search}%"),
31.                     Post.content.ilike(f"%{search}%")
32.                 )
33.             )
34.             .offset(skip)
35.             .limit(limit)
36.         )
37.     else:
38.         query = select(Post).offset(skip).limit(limit)
39.    
40.     result = await db.execute(query)
41.     return result.scalars().all()
43. @router.post("/", response_model=Post)
44. async def create_post(
45.     *,
46.     db: AsyncSession = Depends(get_async_db),
47.     post_in: PostCreate,
48.     # 假设我们有一个异步的get_current_active_user函数
49.     current_user: User = Depends(get_current_active_user),
50. ):
51.     """
52.     异步创建新文章
53.     """
54.     return await async_create_post(db=db, post_in=post_in, owner_id=current_user.id)

复制代码

测试策略

单元测试

使用pytest进行单元测试：

2. # tests/test_crud.py
3. import pytest
4. from sqlalchemy import create_engine
5. from sqlalchemy.orm import sessionmaker
6. from app.db.base_class import Base
7. from app.crud.post import post
8. from app.models.post import Post
9. from app.schemas.post import PostCreate, PostUpdate
10. from app.core.security import get_password_hash
12. # 测试数据库
13. SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
14. engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
15. TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
17. @pytest.fixture
18. def db_session():
19.     Base.metadata.create_all(bind=engine)
20.     db = TestingSessionLocal()
21.     try:
22.         yield db
23.     finally:
24.         db.close()
25.         Base.metadata.drop_all(bind=engine)
27. def test_create_post(db_session):
28.     title = "Test Post"
29.     content = "This is a test post"
30.     post_in = PostCreate(title=title, content=content)
31.    
32.     # 创建一个测试用户
33.     from app.models.user import User
34.     user = User(
35.         email="test@example.com",
36.         hashed_password=get_password_hash("testpassword"),
37.         full_name="Test User",
38.         is_active=True,
39.     )
40.     db_session.add(user)
41.     db_session.commit()
42.     db_session.refresh(user)
43.    
44.     # 创建文章
45.     created_post = post.create_with_owner(db=db_session, obj_in=post_in, owner_id=user.id)
46.    
47.     assert created_post.title == title
48.     assert created_post.content == content
49.     assert created_post.owner_id == user.id
51. def test_get_post(db_session):
52.     title = "Test Post"
53.     content = "This is a test post"
54.     post_in = PostCreate(title=title, content=content)
55.    
56.     # 创建一个测试用户
57.     from app.models.user import User
58.     user = User(
59.         email="test@example.com",
60.         hashed_password=get_password_hash("testpassword"),
61.         full_name="Test User",
62.         is_active=True,
63.     )
64.     db_session.add(user)
65.     db_session.commit()
66.     db_session.refresh(user)
67.    
68.     # 创建文章
69.     created_post = post.create_with_owner(db=db_session, obj_in=post_in, owner_id=user.id)
70.    
71.     # 获取文章
72.     retrieved_post = post.get(db=db_session, id=created_post.id)
73.    
74.     assert retrieved_post
75.     assert retrieved_post.id == created_post.id
76.     assert retrieved_post.title == title
77.     assert retrieved_post.content == content
79. def test_update_post(db_session):
80.     title = "Test Post"
81.     content = "This is a test post"
82.     post_in = PostCreate(title=title, content=content)
83.    
84.     # 创建一个测试用户
85.     from app.models.user import User
86.     user = User(
87.         email="test@example.com",
88.         hashed_password=get_password_hash("testpassword"),
89.         full_name="Test User",
90.         is_active=True,
91.     )
92.     db_session.add(user)
93.     db_session.commit()
94.     db_session.refresh(user)
95.    
96.     # 创建文章
97.     created_post = post.create_with_owner(db=db_session, obj_in=post_in, owner_id=user.id)
98.    
99.     # 更新文章
100.     new_title = "Updated Test Post"
101.     post_update = PostUpdate(title=new_title)
102.     updated_post = post.update(db=db_session, db_obj=created_post, obj_in=post_update)
103.    
104.     assert updated_post.title == new_title
105.     assert updated_post.content == content  # 内容应保持不变

复制代码

API端点测试

使用httpx进行API端点测试：

2. # tests/test_api.py
3. import pytest
4. from fastapi.testclient import TestClient
5. from app.main import app
6. from app.core.config import settings
7. from app.core.security import create_access_token
8. from app.models.user import User
9. from app.schemas.user import UserCreate
11. client = TestClient(app)
13. @pytest.fixture
14. def test_user():
15.     return {
16.         "email": "test@example.com",
17.         "password": "testpassword",
18.         "full_name": "Test User"
19.     }
21. @pytest.fixture
22. def user_token_headers(test_user):
23.     # 创建用户
24.     response = client.post(
25.         f"{settings.API_V1_STR}/users/", json=test_user
26.     )
27.     assert response.status_code == 200
28.    
29.     # 获取访问令牌
30.     login_data = {
31.         "username": test_user["email"],
32.         "password": test_user["password"]
33.     }
34.     response = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
35.     assert response.status_code == 200
36.     token = response.json()["access_token"]
37.    
38.     return {"Authorization": f"Bearer {token}"}
40. def test_create_post(user_token_headers):
41.     post_data = {
42.         "title": "Test Post",
43.         "content": "This is a test post",
44.         "published": False
45.     }
46.     response = client.post(
47.         f"{settings.API_V1_STR}/posts/",
48.         headers=user_token_headers,
49.         json=post_data
50.     )
51.     assert response.status_code == 200
52.     content = response.json()
53.     assert content["title"] == post_data["title"]
54.     assert content["content"] == post_data["content"]
55.     assert content["published"] == post_data["published"]
56.     assert "id" in content
57.     assert "owner_id" in content
59. def test_read_posts(user_token_headers):
60.     # 先创建一个文章
61.     post_data = {
62.         "title": "Test Post for Reading",
63.         "content": "This is a test post for reading",
64.         "published": True
65.     }
66.     client.post(
67.         f"{settings.API_V1_STR}/posts/",
68.         headers=user_token_headers,
69.         json=post_data
70.     )
71.    
72.     # 读取文章列表
73.     response = client.get(
74.         f"{settings.API_V1_STR}/posts/",
75.         headers=user_token_headers
76.     )
77.     assert response.status_code == 200
78.     content = response.json()
79.     assert len(content) >= 1
80.     assert content[0]["title"] == post_data["title"]
82. def test_update_post(user_token_headers):
83.     # 先创建一个文章
84.     post_data = {
85.         "title": "Test Post for Update",
86.         "content": "This is a test post for update",
87.         "published": False
88.     }
89.     response = client.post(
90.         f"{settings.API_V1_STR}/posts/",
91.         headers=user_token_headers,
92.         json=post_data
93.     )
94.     post_id = response.json()["id"]
95.    
96.     # 更新文章
97.     update_data = {
98.         "title": "Updated Test Post",
99.         "published": True
100.     }
101.     response = client.put(
102.         f"{settings.API_V1_STR}/posts/{post_id}",
103.         headers=user_token_headers,
104.         json=update_data
105.     )
106.     assert response.status_code == 200
107.     content = response.json()
108.     assert content["title"] == update_data["title"]
109.     assert content["content"] == post_data["content"]  # 内容应保持不变
110.     assert content["published"] == update_data["published"]
112. def test_delete_post(user_token_headers):
113.     # 先创建一个文章
114.     post_data = {
115.         "title": "Test Post for Deletion",
116.         "content": "This is a test post for deletion",
117.         "published": False
118.     }
119.     response = client.post(
120.         f"{settings.API_V1_STR}/posts/",
121.         headers=user_token_headers,
122.         json=post_data
123.     )
124.     post_id = response.json()["id"]
125.    
126.     # 删除文章
127.     response = client.delete(
128.         f"{settings.API_V1_STR}/posts/{post_id}",
129.         headers=user_token_headers
130.     )
131.     assert response.status_code == 200
132.    
133.     # 验证文章已被删除
134.     response = client.get(
135.         f"{settings.API_V1_STR}/posts/{post_id}",
136.         headers=user_token_headers
137.     )
138.     assert response.status_code == 404

复制代码

部署与监控

Docker部署

使用Docker容器化部署FastAPI应用：

2. # Dockerfile
3. FROM python:3.9-slim
5. # 设置工作目录
6. WORKDIR /app
8. # 设置环境变量
9. ENV PYTHONDONTWRITEBYTECODE 1
10. ENV PYTHONUNBUFFERED 1
12. # 安装系统依赖
13. RUN apt-get update \
14.     && apt-get install -y --no-install-recommends \
15.         gcc \
16.         postgresql-client \
17.         && rm -rf /var/lib/apt/lists/*
19. # 安装Python依赖
20. COPY requirements.txt .
21. RUN pip install --no-cache-dir -r requirements.txt
23. # 复制项目代码
24. COPY . .
26. # 收集静态文件（如果有）
27. # RUN python manage.py collectstatic --noinput
29. # 创建非root用户
30. RUN adduser --disabled-password --gecos '' appuser
31. RUN chown -R appuser:appuser /app
32. USER appuser
34. # 暴露端口
35. EXPOSE 8000
37. # 启动命令
38. CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

复制代码

2. # docker-compose.yml
3. version: '3.8'
5. services:
6.   web:
7.     build: .
8.     ports:
9.       - "8000:8000"
10.     depends_on:
11.       - db
12.       - redis
13.     environment:
14.       - DATABASE_URL=postgresql://fastapi_user:fastapi_password@db:5432/blog_db
15.       - REDIS_URL=redis://redis:6379/0
16.     volumes:
17.       - .:/app
18.     command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
20.   db:
21.     image: postgres:13
22.     environment:
23.       - POSTGRES_USER=fastapi_user
24.       - POSTGRES_PASSWORD=fastapi_password
25.       - POSTGRES_DB=blog_db
26.     volumes:
27.       - postgres_data:/var/lib/postgresql/data/
28.     ports:
29.       - "5432:5432"
31.   redis:
32.     image: redis:6-alpine
33.     ports:
34.       - "6379:6379"
36.   celery:
37.     build: .
38.     command: celery -A app.core.celery_app worker --loglevel=info
39.     volumes:
40.       - .:/app
41.     depends_on:
42.       - db
43.       - redis
44.     environment:
45.       - DATABASE_URL=postgresql://fastapi_user:fastapi_password@db:5432/blog_db
46.       - REDIS_URL=redis://redis:6379/0
47.       - CELERY_BROKER_URL=redis://redis:6379/0
48.       - CELERY_RESULT_BACKEND=redis://redis:6379/0
50.   flower:
51.     build: .
52.     command: celery -A app.core.celery_app flower --port=5555
53.     ports:
54.       - "5555:5555"
55.     depends_on:
56.       - redis
57.     environment:
58.       - CELERY_BROKER_URL=redis://redis:6379/0
59.       - CELERY_RESULT_BACKEND=redis://redis:6379/0
61. volumes:
62.   postgres_data:

复制代码

日志和监控

实现日志记录和监控：

2. # app/core/logging.py
3. import logging
4. import sys
5. from pathlib import Path
6. from app.core.config import settings
8. # 创建日志目录
9. log_dir = Path("logs")
10. log_dir.mkdir(exist_ok=True)
12. # 配置日志
13. def setup_logging():
14.     """
15.     配置应用日志
16.     """
17.     # 创建日志格式化器
18.     formatter = logging.Formatter(
19.         "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
20.     )
21.    
22.     # 创建文件处理器
23.     file_handler = logging.FileHandler("logs/app.log")
24.     file_handler.setFormatter(formatter)
25.     file_handler.setLevel(logging.INFO)
26.    
27.     # 创建控制台处理器
28.     console_handler = logging.StreamHandler(sys.stdout)
29.     console_handler.setFormatter(formatter)
30.     console_handler.setLevel(logging.INFO if not settings.DEBUG else logging.DEBUG)
31.    
32.     # 配置根日志记录器
33.     root_logger = logging.getLogger()
34.     root_logger.setLevel(logging.INFO if not settings.DEBUG else logging.DEBUG)
35.     root_logger.addHandler(file_handler)
36.     root_logger.addHandler(console_handler)
37.    
38.     # 配置第三方库日志级别
39.     logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
40.     logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
41.    
42.     return root_logger

复制代码

2. # app/main.py
3. from fastapi import FastAPI, Request, Response
4. from fastapi.middleware.gzip import GZipMiddleware
5. from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
6. import time
7. import logging
8. from app.api.v1.api import api_router
9. from app.core.config import settings
10. from app.core.logging import setup_logging
12. # 设置日志
13. logger = setup_logging()
15. app = FastAPI(
16.     title=settings.PROJECT_NAME,
17.     openapi_url=f"{settings.API_V1_STR}/openapi.json"
18. )
20. # 添加中间件
21. if settings.ENABLE_HTTPS:
22.     app.add_middleware(HTTPSRedirectMiddleware)
24. app.add_middleware(GZipMiddleware, minimum_size=1000)
26. # 请求日志中间件
27. @app.middleware("http")
28. async def log_requests(request: Request, call_next):
29.     start_time = time.time()
30.    
31.     # 记录请求信息
32.     logger.info(f"Request started: {request.method} {request.url}")
33.    
34.     response = await call_next(request)
35.    
36.     # 计算处理时间
37.     process_time = time.time() - start_time
38.    
39.     # 记录响应信息
40.     logger.info(
41.         f"Request completed: {request.method} {request.url} - Status: {response.status_code} - Time: {process_time:.4f}s"
42.     )
43.    
44.     # 添加处理时间到响应头
45.     response.headers["X-Process-Time"] = str(process_time)
46.    
47.     return response
49. app.include_router(api_router, prefix=settings.API_V1_STR)
51. @app.on_event("startup")
52. async def startup_event():
53.     logger.info("Application startup...")
55. @app.on_event("shutdown")
56. async def shutdown_event():
57.     logger.info("Application shutdown...")

复制代码

性能监控

使用Prometheus和Grafana进行性能监控：

2. # app/core/metrics.py
3. from prometheus_client import Counter, Histogram, Gauge, generate_latest
4. from fastapi import Response
5. import time
7. # 定义指标
8. REQUEST_COUNT = Counter(
9.     'http_requests_total',
10.     'Total HTTP Requests',
11.     ['method', 'endpoint', 'status_code']
12. )
14. REQUEST_DURATION = Histogram(
15.     'http_request_duration_seconds',
16.     'HTTP Request Duration',
17.     ['method', 'endpoint']
18. )
20. ACTIVE_USERS = Gauge(
21.     'active_users',
22.     'Number of active users'
23. )
25. POST_COUNT = Gauge(
26.     'post_count',
27.     'Total number of posts'
28. )
30. # Prometheus端点
31. async def metrics_endpoint():
32.     return Response(generate_latest(), media_type="text/plain")
34. # 指标装饰器
35. def monitor_requests(endpoint_name: str):
36.     def decorator(func):
37.         async def wrapper(*args, **kwargs):
38.             start_time = time.time()
39.             
40.             # 获取请求对象
41.             request = None
42.             for arg in args:
43.                 if hasattr(arg, 'method'):
44.                     request = arg
45.                     break
46.             
47.             # 执行原始函数
48.             response = await func(*args, **kwargs)
49.             
50.             # 计算处理时间
51.             process_time = time.time() - start_time
52.             
53.             # 更新指标
54.             if request:
55.                 REQUEST_COUNT.labels(
56.                     method=request.method,
57.                     endpoint=endpoint_name,
58.                     status_code=response.status_code
59.                 ).inc()
60.                
61.                 REQUEST_DURATION.labels(
62.                     method=request.method,
63.                     endpoint=endpoint_name
64.                 ).observe(process_time)
65.             
66.             return response
67.         
68.         return wrapper
69.     return decorator

复制代码

2. # app/main.py
3. from fastapi import FastAPI
4. from app.api.v1.api import api_router
5. from app.core.config import settings
6. from app.core.metrics import metrics_endpoint
8. app = FastAPI(
9.     title=settings.PROJECT_NAME,
10.     openapi_url=f"{settings.API_V1_STR}/openapi.json"
11. )
13. app.include_router(api_router, prefix=settings.API_V1_STR)
15. # 添加Prometheus指标端点
16. app.add_route("/metrics", metrics_endpoint)

复制代码

总结与最佳实践

通过这个完整的FastAPI实战项目，我们展示了从API设计到系统优化的全过程。以下是一些关键的最佳实践总结：

1. 项目结构设计

• 使用模块化结构，按功能划分代码
• 遵循依赖注入原则，提高代码可测试性
• 使用类型提示，提高代码可读性和IDE支持

2. API设计

• 遵循RESTful设计原则
• 使用Pydantic模型进行数据验证和序列化
• 提供清晰的错误信息和适当的HTTP状态码

3. 认证与授权

• 使用JWT进行无状态认证
• 实现基于角色的访问控制
• 使用FastAPI的依赖注入系统简化权限检查

4. 数据库集成

• 使用SQLAlchemy ORM进行数据库操作
• 实现异步数据库操作提高性能
• 使用数据库迁移工具管理架构变更

5. 缓存策略

• 使用Redis缓存常用数据
• 实现缓存失效机制保持数据一致性
• 使用装饰器简化缓存逻辑

6. 异步处理

• 利用FastAPI的异步特性提高并发性能
• 使用Celery处理后台任务
• 实现任务状态跟踪和结果查询

7. 性能优化

• 优化数据库查询，减少N+1问题
• 使用响应压缩减少传输数据量
• 实现异步端点提高吞吐量

8. 测试策略

• 编写单元测试确保代码质量
• 使用TestClient进行API端点测试
• 实现测试夹具简化测试设置

9. 部署与监控

• 使用Docker容器化应用
• 实现日志记录和监控
• 使用Prometheus和Grafana进行性能监控

通过遵循这些最佳实践，你可以构建出高性能、可维护、可扩展的FastAPI应用，满足现代Web开发的需求。FastAPI的强大功能和简洁设计使其成为构建API的理想选择，无论是小型项目还是大型企业应用，都能胜任。

希望这个实战项目案例能帮助你更好地理解和应用FastAPI，掌握现代Web开发的精髓。

版权声明

1、转载或引用本网站内容(FastAPI实战项目案例分享从API设计到系统优化涵盖认证授权数据库集成缓存策略异步处理等关键技术点助你掌握现代Web开发精髓)须注明原网址及作者(威震华夏关云长)，并标明本网站网址(https://pixtech.cc/)。

2、对于不当转载或引用本网站内容而引起的民事纷争、行政处理或其他损失，本网站不承担责任。

3、对不遵守本声明或其他违法、恶意使用本网站内容者，本网站保留追究其法律责任的权利。

本文地址: https://pixtech.cc/thread-41737-1-1.html