python如何通过FastAPI构建复杂的Web API

通过FastAPI构建复杂的Web API1.项目结构2.依赖管理3.配置管理4.数据库集成5.定义模型与模式6.服务层7.路由分组8.主应用入口9.安全性10.测试11.部署总结FastAPI支持的工具1.依赖注入系统2.Pydantic 模型3.中间件4.异常处理5.背景任务6.静态文件和模板7.WebSocket 支持8.环境变量与配置管理9.测试工具10.OpenAPI 和 Swagger UI11.速率限制FastAPI 路由操作1. GET2. POST3. PUT4. DELETE5. PATCH6. OPTIONS7. HEAD8. TRACE路由参数路径参数查询参数请求体额外功能总结

通过FastAPI构建复杂的Web API

构建复杂的 Web API 通常涉及到多个方面，包括良好的架构设计、清晰的路由组织、数据验证与处理、安全措施、性能优化等。使用 FastAPI 构建复杂 Web API 的关键在于充分利用其提供的工具和特性，同时遵循软件工程的最佳实践。以下是一个详细的指南，帮助你使用 FastAPI 构建一个复杂且高效的 Web API。

1.项目结构

为你的项目创建一个合理的文件夹结构，这有助于代码的组织和维护。例如：

my_project/├── app/│   ├── __init__.py│   ├── main.py│   ├── dependencies.py│   ├── models.py│   ├── schemas.py│   ├── routers/│   │   ├── __init__.py│   │   ├── items.py│   │   └── users.py│   ├── services/│   │   ├── __init__.py│   │   ├── item_service.py│   │   └── user_service.py│   ├── database.py│   └── config.py└── tests/    ├── __init__.py    └── test_main.py

2.依赖管理

使用requirements.txt或者更现代的poetry来管理项目的依赖关系。确保所有开发和生产环境所需的库都被列出。

# 使用 poetry 初始化项目并添加依赖poetry initpoetry add fastapi uvicorn sqlalchemy alembic bcrypt passlib[bcrypt] pydantic email-validator

3.配置管理

使用环境变量或配置文件来管理应用程序的不同设置（如数据库连接字符串、密钥等）。可以借助 Pydantic 的BaseSettings类。

# config.pyfrom pydantic import BaseSettingsclass Settings(BaseSettings):    app_name: str = "My Complex API"    admin_email: str    items_per_user: int = 50    secret_key: str    algorithm: str = "HS256"    access_token_expire_minutes: int = 30    class Config:        env_file = ".env"settings = Settings()

4.数据库集成

选择合适的 ORM（如 SQLAlchemy）并与 FastAPI 集成。创建数据库模型，并考虑使用 Alembic 进行数据库迁移。

# database.pyfrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerSQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)Base = declarative_base()def get_db():    db = SessionLocal()    try:        yield db    finally:        db.close()

5.定义模型与模式

创建数据模型用于映射到数据库表，以及 Pydantic 模型用于请求体和响应的验证。

# models.pyfrom sqlalchemy import Column, Integer, Stringfrom .database import Baseclass Item(Base):    __tablename__ = "items"    id = Column(Integer, primary_key=True, index=True)    name = Column(String, index=True)    description = Column(String, index=True)# schemas.pyfrom pydantic import BaseModelclass ItemCreate(BaseModel):    name: str    description: str = Noneclass Item(BaseModel):    id: int    name: str    description: str = None    class Config:        orm_mode = True

6.服务层

为了保持控制器（路由处理函数）的简洁性，将业务逻辑分离到服务层中。

# services/item_service.pyfrom typing import Listfrom sqlalchemy.orm import Sessionfrom ..models import Item as ItemModelfrom ..schemas import ItemCreate, Itemdef get_items(db: Session, skip: int = 0, limit: int = 10) -> List[Item]:    return db.query(ItemModel).offset(skip).limit(limit).all()def create_item(db: Session, item: ItemCreate) -> Item:    db_item = ItemModel(**item.dict())    db.add(db_item)    db.commit()    db.refresh(db_item)    return db_item

7.路由分组

将相关路径操作分组到不同的路由器模块中，以提高代码的可读性和可维护性。

# routers/items.pyfrom fastapi import APIRouter, Dependsfrom sqlalchemy.orm import Sessionfrom ... import schemas, servicesfrom ...dependencies import get_dbrouter = APIRouter()@router.get("/", response_model=List[schemas.Item])def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):    items = services.get_items(db, skip=skip, limit=limit)    return items@router.post("/", response_model=schemas.Item)def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):    return services.create_item(db=db, item=item)

8.主应用入口

在主应用文件中导入并包含各个路由器。

# main.pyfrom fastapi import FastAPIfrom .routers import items, usersapp = FastAPI()app.include_router(items.router)app.include_router(users.router)

9.安全性

实现身份验证和授权机制，例如使用 JWT Token 进行用户认证。

# dependencies.pyfrom datetime import datetime, timedeltafrom jose import JWTError, jwtfrom fastapi import Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearerfrom . import models, schemas, configoauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")async def get_current_user(token: str = Depends(oauth2_scheme)):    credentials_exception = HTTPException(        status_code=status.HTTP_401_UNAUTHORIZED,        detail="Could not validate credentials",        headers={"WWW-Authenticate": "Bearer"},    )    try:        payload = jwt.decode(token, config.settings.secret_key, algorithms=[config.settings.algorithm])        username: str = payload.get("sub")        if username is None:            raise credentials_exception        token_data = schemas.TokenData(username=username)    except JWTError:        raise credentials_exception    user = get_user(db, username=token_data.username)    if user is None:        raise credentials_exception    return user

10.测试

编写单元测试和集成测试来确保你的 API 按预期工作。FastAPI 提供了方便的测试客户端TestClient。

# tests/test_main.pyfrom fastapi.testclient import TestClientfrom my_project.app.main import appclient = TestClient(app)def test_read_main():    response = client.get("/")    assert response.status_code == 200    assert response.json() == {"message": "Hello World"}

11.部署

考虑使用 Docker 容器化你的应用程序，并通过 CI/CD 管道自动部署到云平台或服务器上。

# DockerfileFROM tiangolo/uvicorn-gunicorn-fastapi:python3.9COPY ./app /appWORKDIR /appRUN pip install --no-cache-dir -r requirements.txtCMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]