Skip to main content Skills Marketplace Descubre y explora habilidades de IA creadas por la comunidad.
Ocupaciones relacionadas SOC
Basado en la clasificación ocupacional SOC
Instalar con Codex o Claude Copia este prompt, pégalo en Codex, Claude u otro asistente, y deja que revise la página de la skill y la instale por ti.
Mostrar detalles del prompt Un comando directo omite el prompt de revisión. Revisa el origen antes de ejecutarlo.
npx skills add https://github.com/tranhieutt/software_development_department --skill fastapi-proEl comando permanece en una sola línea. Desplázate horizontalmente para revisarlo antes de copiarlo.
¿Prefieres una copia local? Descarga los archivos que SkillsMP tiene disponibles ahora.
Descargar Zip Descargando... Más de este repositorio Turns real agent failures, repeated prompts, team-specific workflows, and durable project lessons into better SDD skills or memory entries. Use when the user asks to create/update/refine skills, extract reusable lessons, improve skill routing, encode team process, or save patterns for future sessions.
Provides the vendored agent-style v0.3.5 prose rule pack as a portable Claude skill. Use when installing, syncing, applying, or auditing SDD Agent-Style support in another project, or when a style-review workflow needs the bundled rules.
Reviews Markdown prose against the pinned agent-style v0.3.5 rules and optionally writes a polished copy beside the source. Use for specs, ADRs, PR bodies, release notes, and technical docs when prose quality or evidence discipline matters.
name fastapi-pro type reference description Production FastAPI patterns — async endpoints, SQLAlchemy 2.0 async, Pydantic V2, dependency injection, JWT auth, testing. Use for Python 3.11+ FastAPI backends. NOT for Django (→ `django-patterns`) or Node.js (→ `backend-patterns`). paths ["**/*.py","**/requirements*.txt","**/pyproject.toml","**/main.py","**/app/**/*.py"] effort 3 allowed-tools Read, Glob, Grep, Write, Edit, Bash user-invocable true when_to_use When building async FastAPI APIs with SQLAlchemy 2.0 async + Pydantic V2 — endpoints, auth, DB session handling, testing, deployment
FastAPI Production Patterns
Critical rules (non-obvious)
async def endpoint blocking sync DB callasync DB driver (asyncpg/aiomysql) throughout OR switch endpoint to plain def (FastAPI runs it in threadpool).Pydantic V2 model_config = ConfigDict(...) replaces V1 class Config . Forgetting this silently loses settings like from_attributes=True needed for ORM → DTO conversion.Depends() caches per-requestSQLAlchemy 2.0 async session must not leak across requests : always scope via Depends with async with AsyncSession(...) — raw module-level session causes GreenletError under load.BackgroundTasks runs AFTER response sent in the same worker processUvicorn --workers N forks processes — can't share in-memory state . Use Redis or DB for any shared state (rate-limit counters, cache).
Project layout
app/
├── main.py # FastAPI() instance + lifespan
├── api/
│ ├── deps.py # shared Depends (get_db, get_current_user)
│ └── v1/
│ ├── users.py # APIRouter
│ └── products.py
├── core/
│ ├── config.py # Pydantic Settings
│ ├── security.py # JWT encode/decode, password hashing
│ └── db.py # engine + AsyncSession factory
├── models/ # SQLAlchemy ORM models
├── schemas/ # Pydantic DTOs (Request/Response)
├── services/ # business logic (no framework coupling)
└── tests/
Pydantic V2 settings + config
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings (BaseSettings ):
model_config = SettingsConfigDict(env_file=".env" , env_prefix="APP_"
str
"postgresql+asyncpg://..."
str
32
str
"HS256"
int
30
list
str
list
SQLAlchemy 2.0 async session (per-request)
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
engine = create_async_engine(
settings.database_url,
pool_size=20 ,
max_overflow=10 ,
pool_pre_ping=True ,
echo=False ,
)
SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False )
from typing import AsyncIterator
from fastapi import Depends
async def get_db () -> AsyncIterator[AsyncSession]:
async with SessionLocal() as session:
try :
yield session
await session.commit()
except Exception:
await session.rollback()
raise
Lifespan + startup/shutdown
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan (app: FastAPI ):
async with engine.begin() as conn:
await conn.execute(text("SELECT 1" ))
yield
await engine.dispose()
app = FastAPI(title="My API" , lifespan=lifespan)
JWT auth with OAuth2PasswordBearer
from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from passlib.context import CryptContext
pwd = CryptContext(schemes=["bcrypt" ], deprecated="auto" )
def hash_password (p: str ) -> str : return pwd.hash (p)
def verify_password (p: str , h: str ) -> bool : return pwd.verify(p, h)
def create_access_token (sub: str ) -> str :
exp = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_exp_minutes)
return jwt.encode({"sub" : sub, "exp" : exp}, settings.jwt_secret, settings.jwt_algorithm)
from fastapi import HTTPException, status
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login" )
async def get_current_user (
token: str = Depends(oauth2_scheme ),
db: AsyncSession = Depends(get_db ),
) -> User:
try :
payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
user_id: str = payload.get("sub" )
except JWTError:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid token" )
user = await db.get(User, user_id)
if not user:
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "User not found" )
return user
Endpoint pattern (thin controller, service below)
from fastapi import APIRouter, Depends, HTTPException, status
router = APIRouter(prefix="/users" , tags=["users" ])
@router.post("" , response_model=UserOut, status_code=status.HTTP_201_CREATED
async def create_user (
payload: UserCreate,
db: AsyncSession = Depends(get_db ),
) -> UserOut:
try :
user = await user_service.create(db, payload)
except DuplicateEmailError:
raise HTTPException(status.HTTP_409_CONFLICT, "Email taken" )
return UserOut.model_validate(user)
@router.get("/{user_id}" , response_model=UserOut
async def get_user (
user_id: int ,
db: AsyncSession = Depends(get_db ),
current: User = Depends(get_current_user ),
) -> UserOut:
user = await db.get(User, user_id)
if not user:
raise HTTPException(status.HTTP_404_NOT_FOUND)
if user.id != current.id and not current.is_admin:
raise HTTPException(status.HTTP_403_FORBIDDEN)
return UserOut.model_validate(user)
Pydantic V2 schemas with from_attributes from pydantic import BaseModel, ConfigDict, EmailStr, Field
class UserBase (BaseModel ):
model_config = ConfigDict(from_attributes=True )
email: EmailStr
full_name: str = Field(min_length=1 , max_length=100 )
class UserCreate (UserBase ):
password: str = Field(min_length=8 , max_length=128 )
class UserOut (UserBase ):
id : int
created_at: datetime
Global exception handler
from fastapi import Request
from fastapi.responses import JSONResponse
class AppError (Exception ):
def __init__ (self, msg: str , status_code: int = 400 ):
self .msg, self .status_code = msg, status_code
@app.exception_handler(AppError )
async def app_error_handler (req: Request, exc: AppError ):
return JSONResponse(status_code=exc.status_code, content={"error" : exc.msg})
@app.exception_handler(Exception )
async def unhandled (req: Request, exc: Exception ):
logger.exception("Unhandled error" , extra={"path" : req.url.path})
return JSONResponse(status_code=500 , content={"error" : "Internal server error" })
Testing with pytest-asyncio
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
@pytest_asyncio.fixture
async def client ():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test" ) as c:
yield c
@pytest_asyncio.fixture
async def db_session ():
async with SessionLocal() as s:
yield s
await s.rollback()
@pytest.mark.asyncio
async def test_create_user (client: AsyncClient ):
r = await client.post("/api/v1/users" , json={"email" : "a@b.com" , "full_name" : "A" , "password" : "pw12345678" })
assert r.status_code == 201
assert r.json()["email" ] == "a@b.com"
Production deployment (Uvicorn + Gunicorn)
gunicorn app.main:app \
--workers 4 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 60 \
--keep-alive 5 \
--access-logfile -
Workers = (2 × CPU) + 1 for CPU-bound; lower for IO-heavy async (async workers share event loop already).
Common pitfalls Pitfall Fix async def + sync psycopg2/pymysqlUse asyncpg / aiomysql OR drop async on endpoint BaseSettings V1 pattern (class Config)V2 uses model_config = SettingsConfigDict(...) from_attributes=True missing → validation error from ORM instanceAdd to ConfigDict on every DTO reading from ORM Forgetting await on SQLAlchemy 2.0 async query Linter — use sqlalchemy[asyncio] type stubs + mypy Returning ORM object → leaks relationships (N+1 on serialize) Always model_validate(orm_obj) to scoped Pydantic DTO BackgroundTasks for durable workSwitch to Celery / Dramatiq / ARQ CORSMiddleware added AFTER auth middlewareCORS must be OUTERMOST — added first response_model + return extra fields → silently strippedUse response_model_exclude_unset=True consciously
Observability hooks
import structlog
logger = structlog.get_logger()
@app.middleware("http"
async def log_requests (request: Request, call_next ):
start = time.monotonic()
response = await call_next(request)
logger.info("http_request" ,
method=request.method, path=request.url.path,
status=response.status_code,
duration_ms=(time.monotonic() - start) * 1000 ,
)
return response
Add prometheus-fastapi-instrumentator for /metrics.
Health check: GET /healthz returns {"status": "ok"} + DB SELECT 1.
Request ID: X-Request-ID header middleware → bind to contextvars for logging.
