FastAPI 异步 API 开发教程 - 从基础应用到依赖注入完整指南

python-fastapi-ops by 0xdarkmatter/claude-mods

6 周安装量

16 GitHub Stars

GitHub

安装命令

npx skills add https://github.com/0xdarkmatter/claude-mods --skill python-fastapi-ops

🇨🇳中文介绍

FastAPI 模式

使用 FastAPI 进行现代异步 API 开发。

基础应用

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用程序生命周期 - 启动和关闭。"""
    # 启动
    app.state.db = await create_db_pool()
    yield
    # 关闭
    await app.state.db.close()

app = FastAPI(
    title="My API",
    version="1.0.0",
    lifespan=lifespan,
)

@app.get("/")
async def root():
    return {"message": "Hello World"}

请求/响应模型

from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

class UserCreate(BaseModel):
    """带验证的请求模型。"""
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)

class UserResponse(BaseModel):
    """响应模型。"""
    id: int
    name: str
    email: EmailStr
    created_at: datetime

    model_config = {"from_attributes": True}  # 启用 ORM 模式

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    db_user = await create_user_in_db(user)
    return db_user

路径和查询参数

from fastapi import Query, Path
from typing import Annotated

@app.get("/users/{user_id}")
async def get_user(
    user_id: Annotated[int, Path(..., ge=1, description="User ID")],
):
    return await fetch_user(user_id)

@app.get("/users")
async def list_users(
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 10,
    search: str | None = None,
):
    return await fetch_users(skip=skip, limit=limit, search=search)

🇺🇸English

FastAPI Patterns

Modern async API development with FastAPI.

Basic Application

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan - startup and shutdown."""
    # Startup
    app.state.db = await create_db_pool()
    yield
    # Shutdown
    await app.state.db.close()

app = FastAPI(
    title="My API",
    version="1.0.0",
    lifespan=lifespan,
)

@app.get("/")
async def root():
    return {"message": "Hello World"}

Request/Response Models

from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

class UserCreate(BaseModel):
    """Request model with validation."""
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)

class UserResponse(BaseModel):
    """Response model."""
    id: int
    name: str
    email: EmailStr
    created_at: datetime

    model_config = {"from_attributes": True}  # Enable ORM mode

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    db_user = await create_user_in_db(user)
    return db_user

Path and Query Parameters

广告位招租

在这里展示您的产品或服务

触达数万 AI 开发者，精准高效

联系我们

最新 Skills

AI智能体长期记忆系统 - 精英级架构，融合6种方法，永不丢失上下文

1,200 周安装

AI新闻播客制作技能：实时新闻转对话式播客脚本与音频生成

1,200 周安装

Word文档处理器：DOCX创建、编辑、分析与修订痕迹处理全指南 | 自动化办公解决方案

1,200 周安装

from fastapi import Depends
from typing import Annotated

async def get_db():
    """数据库会话依赖项。"""
    async with async_session() as session:
        yield session

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    """验证并返回当前用户。"""
    user = await authenticate_token(db, token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

# 用于复用的 Annotated 类型
DB = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_user)]

@app.get("/me")
async def get_me(user: CurrentUser):
    return user

from fastapi import HTTPException
from fastapi.responses import JSONResponse

# 内置 HTTP 异常
@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

# 自定义异常处理器
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: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Item {exc.item_id} not found"},
    )

from fastapi import APIRouter

# users.py
router = APIRouter(prefix="/users", tags=["users"])

@router.get("/")
async def list_users():
    return []

@router.get("/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id}

# main.py
from app.routers import users, items

app.include_router(users.router)
app.include_router(items.router, prefix="/api/v1")

功能	用法
路径参数	`@app.get("/items/{id}")`
查询参数	`def f(q: str = None)`
请求体	`def f(item: ItemCreate)`
依赖项	`Depends(get_db)`
认证	`Depends(get_current_user)`
响应模型	`response_model=ItemResponse`
状态码	`status_code=201`

./references/dependency-injection.md - 高级依赖注入模式、作用域、缓存
./references/middleware-patterns.md - 中间件链、CORS、错误处理
./references/validation-serialization.md - Pydantic v2 模式、自定义验证器
./references/background-tasks.md - 后台任务、异步工作器、调度

./scripts/scaffold-api.sh - 生成 API 端点样板代码

./assets/fastapi-template.py - 生产就绪的 FastAPI 应用骨架

python-typing-ops - Pydantic 模型和类型提示
python-async-ops - 异步端点模式

python-database-ops - SQLAlchemy 集成
python-observability-ops - 日志记录、指标、追踪中间件
python-pytest-ops - 使用 TestClient 进行 API 测试

from fastapi import Query, Path
from typing import Annotated

@app.get("/users/{user_id}")
async def get_user(
    user_id: Annotated[int, Path(..., ge=1, description="User ID")],
):
    return await fetch_user(user_id)

@app.get("/users")
async def list_users(
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 10,
    search: str | None = None,
):
    return await fetch_users(skip=skip, limit=limit, search=search)

Dependency Injection

from fastapi import Depends
from typing import Annotated

async def get_db():
    """Database session dependency."""
    async with async_session() as session:
        yield session

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    """Authenticate and return current user."""
    user = await authenticate_token(db, token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

# Annotated types for reuse
DB = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_user)]

@app.get("/me")
async def get_me(user: CurrentUser):
    return user

from fastapi import HTTPException
from fastapi.responses import JSONResponse

# Built-in HTTP exceptions
@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 handler
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: ItemNotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": f"Item {exc.item_id} not found"},
    )

from fastapi import APIRouter

# users.py
router = APIRouter(prefix="/users", tags=["users"])

@router.get("/")
async def list_users():
    return []

@router.get("/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id}

# main.py
from app.routers import users, items

app.include_router(users.router)
app.include_router(items.router, prefix="/api/v1")

Feature	Usage
Path param	`@app.get("/items/{id}")`
Query param	`def f(q: str = None)`
Body	`def f(item: ItemCreate)`
Dependency	`Depends(get_db)`
Auth	`Depends(get_current_user)`
Response model	`response_model=ItemResponse`
Status code	`status_code=201`

Additional Resources

./references/dependency-injection.md - Advanced DI patterns, scopes, caching
./references/middleware-patterns.md - Middleware chains, CORS, error handling
./references/validation-serialization.md - Pydantic v2 patterns, custom validators
./references/background-tasks.md - Background tasks, async workers, scheduling

./scripts/scaffold-api.sh - Generate API endpoint boilerplate

./assets/fastapi-template.py - Production-ready FastAPI app skeleton

python-typing-ops - Pydantic models and type hints
python-async-ops - Async endpoint patterns

python-database-ops - SQLAlchemy integration
python-observability-ops - Logging, metrics, tracing middleware
python-pytest-ops - API testing with TestClient

React Router 框架模式指南：全栈开发、文件路由、数据加载与渲染策略

FastAPI 异步 API 开发教程 - 从基础应用到依赖注入完整指南

🇨🇳中文介绍

FastAPI 模式

基础应用

请求/响应模型

路径和查询参数

🇺🇸English

FastAPI Patterns

Basic Application

Request/Response Models

Path and Query Parameters

相关 Skills

最新 Skills

依赖注入

异常处理

路由器组织

快速参考

额外资源

脚本

资源文件

另请参阅

Dependency Injection

Exception Handling

Router Organization

Quick Reference

Additional Resources

Scripts

Assets

See Also