Back to Details

app-scaffolding

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

FastAPI Skill

FastAPI Skill

Overview

概述

FastAPI is a modern, high-performance async Python web framework for building APIs. It leverages Python type hints and Pydantic for automatic validation, serialization, and OpenAPI documentation generation. FastAPI runs on ASGI servers (Uvicorn, Hypercorn) and provides first-class support for async/await, dependency injection, and WebSockets.
Key characteristics:
  • Automatic interactive API docs (Swagger UI at 
    /docs
    , ReDoc at 
    /redoc
    )
  • Type-driven request validation and response serialization via Pydantic
  • Native async support with full sync fallback
  • Dependency injection system for shared logic and resource management
  • Standards-based: OpenAPI 3.1, JSON Schema
FastAPI是一款用于构建API的现代化、高性能异步Python Web框架。它利用Python类型提示和Pydantic实现自动验证、序列化以及OpenAPI文档生成。FastAPI运行在ASGI服务器（如Uvicorn、Hypercorn）上，为async/await、依赖注入和WebSocket提供一流支持。
核心特性：
  • 自动生成交互式API文档（Swagger UI位于
    /docs
    ，ReDoc位于
    /redoc
    ）
  • 通过Pydantic实现基于类型的请求验证与响应序列化
  • 原生异步支持，同时兼容同步代码
  • 用于共享逻辑与资源管理的依赖注入系统
  • 遵循标准：OpenAPI 3.1、JSON Schema

Project Structure

项目结构

Organize FastAPI projects with a clear separation of concerns:
project/
├── app/
│   ├── __init__.py
│   ├── main.py              # Application factory and lifespan
│   ├── config.py            # Settings via pydantic-settings
│   ├── dependencies.py      # Shared dependencies
│   ├── models.py            # SQLAlchemy / ORM models
│   ├── schemas.py           # Pydantic request/response schemas
│   ├── routers/
│   │   ├── __init__.py      # Router registration
│   │   ├── users.py
│   │   ├── items.py
│   │   └── health.py
│   ├── services/            # Business logic layer
│   ├── middleware/           # Custom middleware
│   └── exceptions.py        # Custom exception handlers
├── tests/
│   ├── conftest.py
│   ├── test_users.py
│   └── test_items.py
├── alembic/                 # Database migrations
├── pyproject.toml
└── .env
See 
assets/app-template/
for a ready-to-use scaffold.
按照关注点分离原则组织FastAPI项目：
project/
├── app/
│   ├── __init__.py
│   ├── main.py              # 应用工厂与生命周期管理
│   ├── config.py            # 基于pydantic-settings的配置
│   ├── dependencies.py      # 共享依赖
│   ├── models.py            # SQLAlchemy / ORM模型
│   ├── schemas.py           # Pydantic请求/响应模型
│   ├── routers/
│   │   ├── __init__.py      # 路由注册
│   │   ├── users.py
│   │   ├── items.py
│   │   └── health.py
│   ├── services/            # 业务逻辑层
│   ├── middleware/           # 自定义中间件
│   └── exceptions.py        # 自定义异常处理器
├── tests/
│   ├── conftest.py
│   ├── test_users.py
│   └── test_items.py
├── alembic/                 # 数据库迁移
├── pyproject.toml
└── .env
可参考
assets/app-template/
获取即用型脚手架。

Application Factory and Lifespan

应用工厂与生命周期

Use the async context manager lifespan pattern (the 
on_event
decorator is deprecated):
python
from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize resources
    app.state.db_pool = await create_db_pool()
    app.state.http_client = httpx.AsyncClient()
    yield
    # Shutdown: release resources
    await app.state.http_client.aclose()
    await app.state.db_pool.close()


def create_app() -> FastAPI:
    app = FastAPI(
        title="My API",
        version="0.1.0",
        lifespan=lifespan,
    )
    app.include_router(users_router)
    app.include_router(items_router)
    return app

app = create_app()
使用异步上下文管理器的生命周期模式（
on_event
装饰器已被弃用）：
python
from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动阶段：初始化资源
    app.state.db_pool = await create_db_pool()
    app.state.http_client = httpx.AsyncClient()
    yield
    # 关闭阶段：释放资源
    await app.state.http_client.aclose()
    await app.state.db_pool.close()


def create_app() -> FastAPI:
    app = FastAPI(
        title="My API",
        version="0.1.0",
        lifespan=lifespan,
    )
    app.include_router(users_router)
    app.include_router(items_router)
    return app

app = create_app()

Route Definition

路由定义

Define routes with HTTP method decorators. Use type-annotated parameters for automatic validation and documentation.
使用HTTP方法装饰器定义路由。通过类型注解参数实现自动验证与文档生成。

Path Parameters

路径参数

python
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> User:
    ...
python
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> User:
    ...

Query Parameters

查询参数

python
@app.get("/items")
async def list_items(skip: int = 0, limit: int = 20, q: str | None = None):
    ...
python
@app.get("/items")
async def list_items(skip: int = 0, limit: int = 20, q: str | None = None):
    ...

Request Body

请求体

python
from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    price: float
    description: str | None = None

@app.post("/items", status_code=201)
async def create_item(item: ItemCreate) -> Item:
    ...
python
from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    price: float
    description: str | None = None

@app.post("/items", status_code=201)
async def create_item(item: ItemCreate) -> Item:
    ...

Multiple Parameter Sources

多参数来源

python
@app.put("/items/{item_id}")
async def update_item(
    item_id: int,                          # path
    item: ItemUpdate,                      # body
    q: str | None = None,                  # query
    x_token: Annotated[str, Header()],     # header
):
    ...
python
@app.put("/items/{item_id}")
async def update_item(
    item_id: int,                          # 路径参数
    item: ItemUpdate,                      # 请求体
    q: str | None = None,                  # 查询参数
    x_token: Annotated[str, Header()],     # 请求头
):
    ...

Response Models

响应模型

Control response shape and status codes:
python
@app.post("/users", response_model=UserOut, status_code=201)
async def create_user(user: UserCreate):
    ...

@app.get("/report", response_class=HTMLResponse)
async def get_report():
    return "<html>...</html>"
Use 
response_model_exclude_unset=True
to omit fields not explicitly set, and 
response_model_exclude={"password"}
to strip sensitive fields.
控制响应结构与状态码：
python
@app.post("/users", response_model=UserOut, status_code=201)
async def create_user(user: UserCreate):
    ...

@app.get("/report", response_class=HTMLResponse)
async def get_report():
    return "<html>...</html>"
使用
response_model_exclude_unset=True
可省略未显式设置的字段，使用
response_model_exclude={"password"}
可移除敏感字段。

Router Organization

路由组织

Split endpoints into routers for modularity:
python
undefined
将端点拆分到不同路由中以实现模块化：
python
undefined

app/routers/users.py

app/routers/users.py

from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/") async def list_users(): ...
@router.get("/{user_id}") async def get_user(user_id: int): ...

Register routers in the application factory:

```python
from app.routers import users, items

app.include_router(users.router)
app.include_router(items.router, prefix="/api/v1")
Apply shared dependencies to all routes on a router:
python
router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(require_admin)],
)
from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/") async def list_users(): ...
@router.get("/{user_id}") async def get_user(user_id: int): ...

在应用工厂中注册路由：

```python
from app.routers import users, items

app.include_router(users.router)
app.include_router(items.router, prefix="/api/v1")
为路由上的所有端点添加共享依赖：
python
router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(require_admin)],
)

Dependency Injection

依赖注入

Use 
Depends()
to declare reusable, composable dependencies. Dependencies can be functions, async functions, or callable classes.
python
from fastapi import Depends
from typing import Annotated

async def get_db():
    async with async_session() as session:
        yield session

DbSession = Annotated[AsyncSession, Depends(get_db)]

async def get_current_user(db: DbSession, token: str = Depends(oauth2_scheme)) -> User:
    ...

CurrentUser = Annotated[User, Depends(get_current_user)]

@app.get("/me")
async def read_me(user: CurrentUser):
    return user
Yield dependencies execute cleanup after the response is sent, making them ideal for database sessions, file handles, and other resources that require teardown.
See 
references/dependency-injection.md
for advanced patterns including parameterized dependencies, class-based dependencies, and testing overrides.
使用
Depends()
声明可复用、可组合的依赖。依赖可以是函数、异步函数或可调用类。
python
from fastapi import Depends
from typing import Annotated

async def get_db():
    async with async_session() as session:
        yield session

DbSession = Annotated[AsyncSession, Depends(get_db)]

async def get_current_user(db: DbSession, token: str = Depends(oauth2_scheme)) -> User:
    ...

CurrentUser = Annotated[User, Depends(get_current_user)]

@app.get("/me")
async def read_me(user: CurrentUser):
    return user
使用yield的依赖会在响应返回后执行清理操作，非常适合数据库会话、文件句柄等需要销毁的资源。
高级模式可参考
references/dependency-injection.md
，包括参数化依赖、基于类的依赖以及测试覆盖。

Request and Response Models with Pydantic

基于Pydantic的请求与响应模型

Define strict, validated schemas using Pydantic models. Separate input and output schemas to control what is accepted vs. returned:
python
class UserBase(BaseModel):
    email: EmailStr
    name: str

class UserCreate(UserBase):
    password: str

class UserOut(UserBase):
    id: int
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)
Cross-reference: see the pydantic skill for comprehensive model patterns, custom validators, and serialization options.
使用Pydantic模型定义严格的验证模式。分离输入与输出模型以控制接收和返回的内容：
python
class UserBase(BaseModel):
    email: EmailStr
    name: str

class UserCreate(UserBase):
    password: str

class UserOut(UserBase):
    id: int
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)
交叉参考：pydantic技能包含全面的模型模式、自定义验证器和序列化选项。

Lifespan Events

生命周期事件

The lifespan async context manager replaces the deprecated 
@app.on_event("startup")
and 
@app.on_event("shutdown")
decorators. Everything before 
yield
runs at startup; everything after runs at shutdown.
Store shared resources on 
app.state
so dependencies can access them:
python
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = await aioredis.from_url("redis://localhost")
    yield
    await app.state.redis.close()
异步上下文管理器的lifespan模式已替代弃用的
@app.on_event("startup")
和
@app.on_event("shutdown")
装饰器。
yield
之前的代码在启动时运行，之后的代码在关闭时运行。
将共享资源存储在
app.state
中，以便依赖可以访问：
python
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = await aioredis.from_url("redis://localhost")
    yield
    await app.state.redis.close()

Background Tasks

后台任务

Enqueue lightweight work to run after the response is returned:
python
from fastapi import BackgroundTasks

def send_notification(email: str, message: str):
    # blocking or async work
    ...

@app.post("/orders")
async def create_order(order: OrderCreate, background_tasks: BackgroundTasks):
    result = await process_order(order)
    background_tasks.add_task(send_notification, order.email, "Order confirmed")
    return result
For heavier workloads, offload to a task queue (Celery, ARQ, or Dramatiq) instead.
将轻量型工作加入队列，在响应返回后执行：
python
from fastapi import BackgroundTasks

def send_notification(email: str, message: str):
    # 阻塞或异步任务
    ...

@app.post("/orders")
async def create_order(order: OrderCreate, background_tasks: BackgroundTasks):
    result = await process_order(order)
    background_tasks.add_task(send_notification, order.email, "Order confirmed")
    return result
对于更繁重的工作负载，建议将其卸载到任务队列（如Celery、ARQ或Dramatiq）。

Error Handling

错误处理

HTTPException

HTTPException

python
from fastapi import HTTPException

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    item = await fetch_item(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item
python
from fastapi import HTTPException

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    item = await fetch_item(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

Custom Exception Handlers

自定义异常处理器

python
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request, exc):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Item {exc.item_id} not found"},
    )

@app.exception_handler(RequestValidationError)
async def validation_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body},
    )
python
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request, exc):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Item {exc.item_id} not found"},
    )

@app.exception_handler(RequestValidationError)
async def validation_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body},
    )

CORS Configuration

CORS配置

python
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
Never use 
allow_origins=["*"]
with 
allow_credentials=True
in production. See 
references/middleware.md
for detailed CORS guidance and other middleware patterns.
python
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
生产环境中切勿同时使用
allow_origins=["*"]
和
allow_credentials=True
。详细的CORS指南及其他中间件模式可参考
references/middleware.md
。

Testing

测试

Use 
TestClient
for synchronous tests or 
httpx.AsyncClient
for async tests. Override dependencies to inject mocks:
python
from fastapi.testclient import TestClient

def test_read_items():
    app.dependency_overrides[get_db] = lambda: mock_db
    client = TestClient(app)
    response = client.get("/items")
    assert response.status_code == 200
    app.dependency_overrides.clear()
Async test with 
httpx
:
python
import pytest
from httpx import ASGITransport, AsyncClient

@pytest.mark.anyio
async def test_create_item():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.post("/items", json={"name": "Widget", "price": 9.99})
    assert response.status_code == 201
使用
TestClient
进行同步测试，或使用
httpx.AsyncClient
进行异步测试。覆盖依赖以注入模拟对象：
python
from fastapi.testclient import TestClient

def test_read_items():
    app.dependency_overrides[get_db] = lambda: mock_db
    client = TestClient(app)
    response = client.get("/items")
    assert response.status_code == 200
    app.dependency_overrides.clear()
使用
httpx
进行异步测试：
python
import pytest
from httpx import ASGITransport, AsyncClient

@pytest.mark.anyio
async def test_create_item():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as client:
        response = await client.post("/items", json={"name": "Widget", "price": 9.99})
    assert response.status_code == 201

Further Reading

延伸阅读

  • references/routing-patterns.md
    -- advanced routing, file uploads, WebSockets, SSE
  • references/middleware.md
    -- middleware patterns, ordering, custom middleware
  • references/dependency-injection.md
    -- DI composition, testing, scoping
  • assets/app-template/
    -- production-ready project scaffold
Cross-references:
  • pydantic skill -- model definitions, validators, serialization
  • async-patterns skill -- async/await best practices, concurrency
  • uvicorn skill -- ASGI server configuration and deployment
  • references/routing-patterns.md
    -- 高级路由、文件上传、WebSocket、SSE
  • references/middleware.md
    -- 中间件模式、执行顺序、自定义中间件
  • references/dependency-injection.md
    -- 依赖注入组合、测试、作用域
  • assets/app-template/
    -- 生产级项目脚手架
交叉参考：
  • pydantic技能 -- 模型定义、验证器、序列化
  • async-patterns技能 -- async/await最佳实践、并发
  • uvicorn技能 -- ASGI服务器配置与部署