菜单
Use when building FastAPI applications, implementing async endpoints, setting up Pydantic schemas, working with SQLAlchemy, or writing pytest tests for Python backend services.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/MadAppGang/claude-code --skill python命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
基于 SOC 职业分类
Use when starting isolated feature work or before executing implementation plans. Manages full worktree lifecycle from creation through cleanup with safety checks and error recovery.
Configuration reference and troubleshooting for the statusline plugin — sections, themes, bar widths, and script architecture
Common agent patterns and templates for Claude Code. Use when implementing agents to follow proven patterns for Tasks integration, quality checks, and external model invocation via claudish CLI.
正在显示 SKILL.md
| name | python |
| version | 1.0.0 |
| description | Use when building FastAPI applications, implementing async endpoints, setting up Pydantic schemas, working with SQLAlchemy, or writing pytest tests for Python backend services. |
| keywords | ["Python","FastAPI","Pydantic","SQLAlchemy","async","pytest","backend","API"] |
| plugin | dev |
| updated | "2026-01-20T00:00:00.000Z" |
Python patterns for building backend services with FastAPI.
project/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI app
│ ├── config.py # Configuration
│ ├── dependencies.py # Dependency injection
│ ├── routers/ # API routes
│ │ ├── __init__.py
│ │ └── users.py
│ ├── services/ # Business logic
│ ├── repositories/ # Data access
│ ├── models/ # SQLAlchemy models
│ ├── schemas/ # Pydantic schemas
│ └── utils/ # Utilities
├── tests/ # Test files
├── migrations/ # Alembic migrations
├── pyproject.toml
└── requirements.txt
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from app.config import settings
from app.routers import users, auth
from app.database import engine, Base
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
# Shutdown
await engine.dispose()
app = FastAPI(
title="My API",
version="1.0.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/users", tags=["users"])
@app.get("/health")
async def health():
return {"status": "ok"}
# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
database_url: str = "postgresql+asyncpg://localhost/app"
redis_url: str = "redis://localhost:6379"
secret_key: str = "your-secret-key"
access_token_expire_minutes: int = 30
cors_origins: list[str] = ["http://localhost:3000"]
class Config:
env_file = ".env"
@lru_cache
def get_settings() -> Settings:
return Settings()
settings = get_settings()
# app/schemas/user.py
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Optional
class UserBase(BaseModel):
name: str = Field(..., min_length=2, max_length=100)
email: EmailStr
class UserCreate(UserBase):
password: str = Field(..., min_length=8)
class UserUpdate(BaseModel):
name: Optional[str] = Field(None, min_length=2, max_length=100)
email: Optional[EmailStr] = None
class UserResponse(UserBase):
id: str
created_at: datetime
class Config:
from_attributes = True
class PaginatedResponse(BaseModel):
items: list[UserResponse]
total: int
page: int
page_size: int
# app/models/user.py
from sqlalchemy import Column, String, DateTime, Boolean
from sqlalchemy.sql import func
from app.database import Base
import uuid
class User(Base):
__tablename__ = "users"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
name = Column(String(100), nullable=False)
email = Column(String(255), unique=True, nullable=False, index=True)
password_hash = Column(String(255), nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from app.config import settings
engine = create_async_engine(settings.database_url, echo=True)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
Base = declarative_base()
async def get_db():
async with async_session() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
finally:
await session.close()
# app/repositories/user.py
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
class UserRepository:
def __init__(self, session: AsyncSession):
self.session = session
async def find_by_id(self, id: str) -> User | None:
result = await self.session.execute(
select(User).where(User.id == id)
)
return result.scalar_one_or_none()
async def find_by_email(self, email: str) -> User | None:
result = await self.session.execute(
select(User).where(User.email == email)
)
return result.scalar_one_or_none()
async def find_all(self, page: int = 1, page_size: int = 20) -> tuple[list[User], int]:
offset = (page - 1) * page_size
# Get items
result = await self.session.execute(
select(User)
.order_by(User.created_at.desc())
.offset(offset)
.limit(page_size)
)
items = result.scalars().all()
# Get total count
count_result = await self.session.execute(
select(func.count()).select_from(User)
)
total = count_result.scalar()
return list(items), total
async def create(self, data: UserCreate, password_hash: str) -> User:
user = User(
name=data.name,
email=data.email,
password_hash=password_hash,
)
self.session.add(user)
await self.session.flush()
return user
async def update(self, user: User, data: UserUpdate) -> User:
for key, value in data.model_dump(exclude_unset=True).items():
setattr(user, key, value)
await self.session.flush()
return user
async def delete(self, user: User) -> None:
await self.session.delete(user)
# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.repositories.user import UserRepository
from app.services.user import UserService
from app.schemas.user import UserCreate, UserUpdate, UserResponse, PaginatedResponse
from app.dependencies import get_current_user
router = APIRouter()
def get_user_service(db: AsyncSession = Depends(get_db)) -> UserService:
return UserService(UserRepository(db))
@router.get("", response_model=PaginatedResponse)
async def list_users(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
service: UserService = Depends(get_user_service),
):
users, total = await service.find_all(page, page_size)
return PaginatedResponse(
items=users,
total=total,
page=page,
page_size=page_size,
)
@router.get("/{id}", response_model=UserResponse)
async def get_user(
id: str,
service: UserService = Depends(get_user_service),
):
user = await service.find_by_id(id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
data: UserCreate,
service: UserService = Depends(get_user_service),
):
return await service.create(data)
@router.patch("/{id}", response_model=UserResponse)
async def update_user(
id: str,
data: UserUpdate,
service: UserService = Depends(get_user_service),
current_user: User = Depends(get_current_user),
):
user = await service.update(id, data)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
id: str,
service: UserService = Depends(get_user_service),
current_user: User = Depends(get_current_user),
):
success = await service.delete(id)
if not success:
raise HTTPException(status_code=404, detail="User not found")
# app/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from app.config import settings
from app.models.user import User
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db),
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
user_id: str = payload.get("sub")
if user_id is None:
raise credentials_exception
except JWTError:
raise credentials_exception
repo = UserRepository(db)
user = await repo.find_by_id(user_id)
if user is None:
raise credentials_exception
return user
# tests/test_users.py
import pytest
from httpx import AsyncClient
from app.main import app
from app.database import get_db, async_session
@pytest.fixture
async def client():
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.fixture
async def db():
async with async_session() as session:
yield session
await session.rollback()
@pytest.mark.asyncio
async def test_list_users(client: AsyncClient):
response = await client.get("/api/users")
assert response.status_code == 200
data = response.json()
assert "items" in data
assert "total" in data
@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
response = await client.post("/api/users", json={
"name": "John Doe",
"email": "john@example.com",
"password": "password123",
})
assert response.status_code == 201
data = response.json()
assert data["name"] == "John Doe"
assert data["email"] == "john@example.com"
@pytest.mark.asyncio
async def test_create_user_validation(client: AsyncClient):
response = await client.post("/api/users", json={
"name": "J", # Too short
"email": "invalid", # Invalid email
})
assert response.status_code == 422
Python FastAPI patterns for backend development