Back to Details

python-backend

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

python-backend

Python后端开发

Production-ready Python backend patterns for FastAPI, SQLAlchemy, and Upstash.
适用于FastAPI、SQLAlchemy和Upstash的生产级Python后端实践模式。

When to Use This Skill

适用场景

  • Building REST APIs with FastAPI
  • Implementing JWT/OAuth2 authentication
  • Setting up SQLAlchemy async databases
  • Integrating Redis/Upstash caching and rate limiting
  • Refactoring AI-generated Python code
  • Designing API patterns and project structure
  • 使用FastAPI构建REST API
  • 实现JWT/OAuth2认证
  • 搭建SQLAlchemy异步数据库
  • 集成Redis/Upstash缓存与限流
  • 重构AI生成的Python代码
  • 设计API模式与项目结构

Core Principles

核心原则

  1. Async-first - Use async/await for I/O operations
  2. Type everything - Pydantic models for validation
  3. Dependency injection - Use FastAPI's Depends()
  4. Fail fast - Validate early, use HTTPException
  5. Security by default - Never trust user input
  1. 优先异步 - 对I/O操作使用async/await
  2. 全类型标注 - 使用Pydantic模型进行验证
  3. 依赖注入 - 使用FastAPI的Depends()
  4. 快速失败 - 尽早验证，使用HTTPException
  5. 默认安全 - 绝不信任用户输入

Quick Patterns

快速实践模式

Project Structure

项目结构

src/
├── auth/
│   ├── router.py      # endpoints
│   ├── schemas.py     # pydantic models
│   ├── models.py      # db models
│   ├── service.py     # business logic
│   └── dependencies.py
├── posts/
│   └── ...
├── config.py
├── database.py
└── main.py
src/
├── auth/
│   ├── router.py      # 接口端点
│   ├── schemas.py     # Pydantic模型
│   ├── models.py      # 数据库模型
│   ├── service.py     # 业务逻辑
│   └── dependencies.py
├── posts/
│   └── ...
├── config.py
├── database.py
└── main.py

Async Routes

异步路由

python
undefined
python
undefined

BAD - blocks event loop

错误示例 - 阻塞事件循环

@router.get("/") async def bad(): time.sleep(10) # Blocking!
@router.get("/") async def bad(): time.sleep(10) # 阻塞操作！

GOOD - runs in threadpool

良好示例 - 在线程池中运行

@router.get("/") def good(): time.sleep(10) # OK in sync function
@router.get("/") def good(): time.sleep(10) # 在同步函数中允许

BEST - non-blocking

最佳实践 - 非阻塞

@router.get("/") async def best(): await asyncio.sleep(10) # Non-blocking
undefined
@router.get("/") async def best(): await asyncio.sleep(10) # 非阻塞操作
undefined

Pydantic Validation

Pydantic验证

python
from pydantic import BaseModel, EmailStr, Field

class UserCreate(BaseModel):
    email: EmailStr
    username: str = Field(min_length=3, max_length=50, pattern="^[a-zA-Z0-9_]+$")
    age: int = Field(ge=18)
python
from pydantic import BaseModel, EmailStr, Field

class UserCreate(BaseModel):
    email: EmailStr
    username: str = Field(min_length=3, max_length=50, pattern="^[a-zA-Z0-9_]+$")
    age: int = Field(ge=18)

Dependency Injection

依赖注入

python
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    payload = decode_token(token)
    user = await get_user(payload["sub"])
    if not user:
        raise HTTPException(401, "User not found")
    return user

@router.get("/me")
async def get_me(user: User = Depends(get_current_user)):
    return user
python
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    payload = decode_token(token)
    user = await get_user(payload["sub"])
    if not user:
        raise HTTPException(401, "用户不存在")
    return user

@router.get("/me")
async def get_me(user: User = Depends(get_current_user)):
    return user

SQLAlchemy Async

SQLAlchemy异步实现

python
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session
python
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        yield session

Redis Caching

Redis缓存

python
from upstash_redis import Redis

redis = Redis.from_env()

@app.get("/data/{id}")
def get_data(id: str):
    cached = redis.get(f"data:{id}")
    if cached:
        return cached
    data = fetch_from_db(id)
    redis.setex(f"data:{id}", 600, data)
    return data
python
from upstash_redis import Redis

redis = Redis.from_env()

@app.get("/data/{id}")
def get_data(id: str):
    cached = redis.get(f"data:{id}")
    if cached:
        return cached
    data = fetch_from_db(id)
    redis.setex(f"data:{id}", 600, data)
    return data

Rate Limiting

限流

python
from upstash_ratelimit import Ratelimit, SlidingWindow

ratelimit = Ratelimit(
    redis=Redis.from_env(),
    limiter=SlidingWindow(max_requests=10, window=60),
)

@app.get("/api/resource")
def protected(request: Request):
    result = ratelimit.limit(request.client.host)
    if not result.allowed:
        raise HTTPException(429, "Rate limit exceeded")
    return {"data": "..."}
python
from upstash_ratelimit import Ratelimit, SlidingWindow

ratelimit = Ratelimit(
    redis=Redis.from_env(),
    limiter=SlidingWindow(max_requests=10, window=60),
)

@app.get("/api/resource")
def protected(request: Request):
    result = ratelimit.limit(request.client.host)
    if not result.allowed:
        raise HTTPException(429, "请求频率超限")
    return {"data": "..."}

Reference Documents

参考文档

For detailed patterns, see:
DocumentContent
references/fastapi_patterns.md
Project structure, async, Pydantic, dependencies, testing
references/security_patterns.md
JWT, OAuth2, password hashing, CORS, API keys
references/database_patterns.md
SQLAlchemy async, transactions, eager loading, migrations
references/upstash_patterns.md
Redis, rate limiting, QStash background jobs
如需详细实践模式，请参阅：
文档内容
references/fastapi_patterns.md
项目结构、异步实现、Pydantic、依赖注入、测试
references/security_patterns.md
JWT、OAuth2、密码哈希、CORS、API密钥
references/database_patterns.md
SQLAlchemy异步、事务处理、预加载、迁移
references/upstash_patterns.md
Redis、限流、QStash后台任务

Resources

资源