python-fastapi-ops by 0xdarkmatter/claude-mods
npx skills add https://github.com/0xdarkmatter/claude-mods --skill python-fastapi-ops使用 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)
Modern async API development with FastAPI.
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"}
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
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
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 测试每周安装数
1
代码仓库
GitHub 星标数
8
首次出现
1 天前
安全审计
安装于
zencoder1
amp1
cline1
openclaw1
opencode1
cursor1
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)
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 |
./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 skeletonPrerequisites:
python-typing-ops - Pydantic models and type hintspython-async-ops - Async endpoint patternsRelated Skills:
python-database-ops - SQLAlchemy integrationpython-observability-ops - Logging, metrics, tracing middlewarepython-pytest-ops - API testing with TestClientWeekly Installs
1
Repository
GitHub Stars
8
First Seen
1 day ago
Security Audits
Installed on
zencoder1
amp1
cline1
openclaw1
opencode1
cursor1
React Router 框架模式指南:全栈开发、文件路由、数据加载与渲染策略
1,200 周安装