关于 python-backend
python-backend 为使用 FastAPI、SQLAlchemy 和 Upstash 构建 Python 后端的开发者提供参考技能,涵盖 API 结构、异步数据访问、验证、身份认证、缓存和限流模式。
主要功能
- 01FastAPI 项目与 API 模式
- 02SQLAlchemy 异步数据库访问
- 03Pydantic 验证与依赖注入
- 04JWT 和 OAuth2 安全模式
- 05Upstash 缓存与限流
适用场景
- 01规划 FastAPI REST API 结构
- 02添加异步数据库访问
- 03实现身份认证和输入验证
- 04设置缓存或请求限流
安装
复制下面的命令,在终端中运行。
$
npx skills add https:/ / github.com/ jiatastic/ open-python-skills --skill python-backend安装后在 Claude Code 中输入/python-backend确认已加载。
包含的文件
除 SKILL.md 外,安装时一起下载的参考资料和脚本。
references/database_patterns.mdfastapi_patterns.mdsecurity_patterns.mdupstash_patterns.md详细介绍
python-backend
适用于 FastAPI、SQLAlchemy 和 Upstash 的生产就绪 Python 后端模式。
何时使用此技能
- 使用 FastAPI 构建 REST API
- 实现 JWT/OAuth2 身份验证
- 设置 SQLAlchemy 异步数据库
- 集成 Redis/Upstash 缓存和速率限制
- 重构 AI 生成的 Python 代码
- 设计 API 模式和项目结构
核心原则
- 异步优先 - 对 I/O 操作使用 async/await
- 为所有内容添加类型 - 使用 Pydantic 模型进行验证
- 依赖注入 - 使用 FastAPI 的 Depends()
- 快速失败 - 尽早验证,使用 HTTPException
- 默认安全 - 绝不信任用户输入
快速模式
项目结构
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
异步路由
# BAD - blocks event loop
@router.get("/")
async def bad():
time.sleep(10) # Blocking!
# GOOD - runs in threadpool
@router.get("/")
def good():
time.sleep(10) # OK in sync function
# BEST - non-blocking
@router.get("/")
async def best():
await asyncio.sleep(10) # Non-blocking
Pydantic 验证
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)
依赖注入
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
SQLAlchemy 异步
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 缓存
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
速率限制
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": "..."}
参考文档
如需了解详细模式,请参阅:
| 文档 | 内容 |
|---|---|
references/fastapi_patterns.md | 项目结构、异步、Pydantic、依赖项、测试 |
references/security_patterns.md | JWT、OAuth2、密码哈希、CORS、API 密钥 |
references/database_patterns.md | SQLAlchemy 异步、事务、预加载、迁移 |
references/upstash_patterns.md | Redis、速率限制、QStash 后台任务 |
资源
评论
支持 Markdown