mirror of
https://github.com/duthaho/claudekit.git
synced 2026-07-17 21:19:41 +03:00
feat: adding new skills, including testing patterns and methodologies, along with bundled resources for better usability.
This commit is contained in:
duthaho
@@ -1,89 +1,680 @@
|
||||
---
|
||||
name: fastapi
|
||||
description: >
|
||||
Use this skill when building REST APIs with Python and FastAPI, creating async web applications, or generating OpenAPI/Swagger documentation. Trigger for any mention of FastAPI, Pydantic models, async Python endpoints, dependency injection in Python APIs, or APIRouter patterns. Also applies when setting up Python microservices, adding request validation with Pydantic, or configuring ASGI applications.
|
||||
---
|
||||
|
||||
# FastAPI
|
||||
|
||||
## Description
|
||||
|
||||
FastAPI web framework with async patterns, Pydantic validation, and OpenAPI documentation.
|
||||
|
||||
## When to Use
|
||||
|
||||
- Building REST APIs with Python
|
||||
- Async web applications
|
||||
- OpenAPI/Swagger documentation needed
|
||||
- Python microservices
|
||||
- WebSocket real-time applications
|
||||
|
||||
## When NOT to Use
|
||||
|
||||
- Django projects — use the `frameworks/django` skill instead
|
||||
- JavaScript/Node.js backends (Express, NestJS) — this skill is Python-only
|
||||
- Non-API applications such as CLI tools, desktop apps, or batch processing scripts
|
||||
|
||||
---
|
||||
|
||||
## Core Patterns
|
||||
|
||||
### Route Definition
|
||||
### 1. Project Structure
|
||||
|
||||
Recommended layout for medium-large FastAPI applications:
|
||||
|
||||
```
|
||||
project/
|
||||
├── app/
|
||||
│ ├── __init__.py
|
||||
│ ├── main.py # FastAPI app creation, startup/shutdown
|
||||
│ ├── config.py # Settings via pydantic-settings
|
||||
│ ├── dependencies.py # Shared dependencies
|
||||
│ ├── exceptions.py # Custom exception handlers
|
||||
│ ├── middleware.py # Custom middleware
|
||||
│ ├── api/
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── router.py # Root router aggregating all sub-routers
|
||||
│ │ ├── v1/
|
||||
│ │ │ ├── __init__.py
|
||||
│ │ │ ├── users.py # /api/v1/users endpoints
|
||||
│ │ │ ├── items.py # /api/v1/items endpoints
|
||||
│ │ │ └── auth.py # /api/v1/auth endpoints
|
||||
│ │ └── v2/ # Future API version
|
||||
│ ├── models/
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── user.py # SQLAlchemy / SQLModel ORM models
|
||||
│ │ └── item.py
|
||||
│ ├── schemas/
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── user.py # Pydantic request/response schemas
|
||||
│ │ └── item.py
|
||||
│ ├── services/
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── user_service.py # Business logic layer
|
||||
│ │ └── item_service.py
|
||||
│ ├── repositories/
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── user_repo.py # Data access layer
|
||||
│ │ └── item_repo.py
|
||||
│ ├── core/
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── database.py # DB engine, session factory
|
||||
│ │ └── security.py # JWT, hashing, auth utils
|
||||
│ └── tests/
|
||||
│ ├── __init__.py
|
||||
│ ├── conftest.py # Fixtures: test client, test DB
|
||||
│ ├── test_users.py
|
||||
│ └── test_items.py
|
||||
├── alembic/ # Database migrations
|
||||
│ ├── env.py
|
||||
│ └── versions/
|
||||
├── alembic.ini
|
||||
├── pyproject.toml
|
||||
├── Dockerfile
|
||||
└── .env
|
||||
```
|
||||
|
||||
**Key conventions:**
|
||||
- Separate `schemas/` (Pydantic) from `models/` (ORM) to keep concerns clean
|
||||
- Use `services/` for business logic, `repositories/` for data access
|
||||
- Version API routes under `api/v1/`, `api/v2/` for backward compatibility
|
||||
- Keep `main.py` thin — it only wires things together
|
||||
|
||||
```python
|
||||
from fastapi import FastAPI, HTTPException, Depends
|
||||
from pydantic import BaseModel
|
||||
# app/main.py
|
||||
from contextlib import asynccontextmanager
|
||||
from fastapi import FastAPI
|
||||
from app.api.router import api_router
|
||||
from app.config import settings
|
||||
from app.core.database import engine
|
||||
from app.middleware import add_middleware
|
||||
|
||||
app = FastAPI()
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
# Startup: create tables, warm caches, connect to services
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
yield
|
||||
# Shutdown: close connections, flush buffers
|
||||
await engine.dispose()
|
||||
|
||||
class UserCreate(BaseModel):
|
||||
email: str
|
||||
name: str
|
||||
app = FastAPI(
|
||||
title=settings.PROJECT_NAME,
|
||||
version=settings.VERSION,
|
||||
lifespan=lifespan,
|
||||
)
|
||||
add_middleware(app)
|
||||
app.include_router(api_router, prefix="/api")
|
||||
```
|
||||
|
||||
class UserResponse(BaseModel):
|
||||
id: int
|
||||
email: str
|
||||
name: str
|
||||
### 2. Route Patterns
|
||||
|
||||
@app.post("/users", response_model=UserResponse, status_code=201)
|
||||
async def create_user(user: UserCreate):
|
||||
# Create user logic
|
||||
return UserResponse(id=1, **user.model_dump())
|
||||
#### APIRouter with tags, prefixes, and dependencies
|
||||
|
||||
@app.get("/users/{user_id}", response_model=UserResponse)
|
||||
async def get_user(user_id: int):
|
||||
user = await get_user_by_id(user_id)
|
||||
```python
|
||||
from fastapi import APIRouter, Depends, Query, Path, Body, HTTPException, status
|
||||
from app.schemas.user import UserCreate, UserResponse, UserUpdate, UserList
|
||||
from app.dependencies import get_current_user
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/users",
|
||||
tags=["users"],
|
||||
dependencies=[Depends(get_current_user)], # Applied to all routes
|
||||
responses={401: {"description": "Not authenticated"}},
|
||||
)
|
||||
|
||||
# Path parameters with validation
|
||||
@router.get("/{user_id}", response_model=UserResponse)
|
||||
async def get_user(
|
||||
user_id: int = Path(..., gt=0, description="The ID of the user to retrieve"),
|
||||
):
|
||||
user = await user_service.get(user_id)
|
||||
if not user:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return user
|
||||
|
||||
# Query parameters with defaults and validation
|
||||
@router.get("/", response_model=UserList)
|
||||
async def list_users(
|
||||
skip: int = Query(0, ge=0, description="Number of records to skip"),
|
||||
limit: int = Query(20, ge=1, le=100, description="Max records to return"),
|
||||
search: str | None = Query(None, min_length=1, max_length=100),
|
||||
sort_by: str = Query("created_at", pattern="^(created_at|name|email)$"),
|
||||
):
|
||||
users = await user_service.list(skip=skip, limit=limit, search=search)
|
||||
return users
|
||||
|
||||
# Request body with status codes
|
||||
@router.post(
|
||||
"/",
|
||||
response_model=UserResponse,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
summary="Create a new user",
|
||||
description="Creates a user account and sends a welcome email.",
|
||||
)
|
||||
async def create_user(user: UserCreate = Body(...)):
|
||||
return await user_service.create(user)
|
||||
|
||||
# Multiple response models for different status codes
|
||||
@router.put("/{user_id}", response_model=UserResponse, responses={
|
||||
404: {"description": "User not found"},
|
||||
409: {"description": "Email already taken"},
|
||||
})
|
||||
async def update_user(user_id: int, user: UserUpdate):
|
||||
return await user_service.update(user_id, user)
|
||||
|
||||
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_user(user_id: int):
|
||||
await user_service.delete(user_id)
|
||||
```
|
||||
|
||||
### Dependency Injection
|
||||
#### Router aggregation
|
||||
|
||||
```python
|
||||
from fastapi import Depends
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
# app/api/router.py
|
||||
from fastapi import APIRouter
|
||||
from app.api.v1 import users, items, auth
|
||||
|
||||
api_router = APIRouter()
|
||||
api_router.include_router(auth.router, prefix="/v1")
|
||||
api_router.include_router(users.router, prefix="/v1")
|
||||
api_router.include_router(items.router, prefix="/v1")
|
||||
```
|
||||
|
||||
### 3. Dependency Injection
|
||||
|
||||
#### Basic dependency with Depends()
|
||||
|
||||
```python
|
||||
from fastapi import Depends, Header, HTTPException
|
||||
|
||||
async def verify_api_key(x_api_key: str = Header(...)):
|
||||
if x_api_key != settings.API_KEY:
|
||||
raise HTTPException(status_code=403, detail="Invalid API key")
|
||||
return x_api_key
|
||||
```
|
||||
|
||||
#### Nested dependencies
|
||||
|
||||
```python
|
||||
async def get_db() -> AsyncGenerator[AsyncSession, None]:
|
||||
async with async_session_maker() as session:
|
||||
yield session
|
||||
yield session # yield dependency — cleanup runs after response
|
||||
|
||||
@app.get("/users")
|
||||
async def list_users(db: AsyncSession = Depends(get_db)):
|
||||
return await db.execute(select(User))
|
||||
async def get_user_repo(db: AsyncSession = Depends(get_db)) -> UserRepository:
|
||||
return UserRepository(db)
|
||||
|
||||
async def get_user_service(
|
||||
repo: UserRepository = Depends(get_user_repo),
|
||||
) -> UserService:
|
||||
return UserService(repo)
|
||||
|
||||
@router.get("/users")
|
||||
async def list_users(service: UserService = Depends(get_user_service)):
|
||||
return await service.list_all()
|
||||
```
|
||||
|
||||
### Router Organization
|
||||
#### Yield dependencies for cleanup
|
||||
|
||||
```python
|
||||
from fastapi import APIRouter
|
||||
async def get_redis() -> AsyncGenerator[Redis, None]:
|
||||
redis = await aioredis.from_url(settings.REDIS_URL)
|
||||
try:
|
||||
yield redis
|
||||
finally:
|
||||
await redis.close() # Always runs, even on exceptions
|
||||
|
||||
router = APIRouter(prefix="/users", tags=["users"])
|
||||
|
||||
@router.get("/")
|
||||
async def list_users():
|
||||
pass
|
||||
|
||||
# In main.py
|
||||
app.include_router(router)
|
||||
async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
yield client
|
||||
```
|
||||
|
||||
#### Request-scoped dependencies with caching
|
||||
|
||||
```python
|
||||
# FastAPI caches dependency results per-request by default.
|
||||
# The same db session is reused if multiple deps request it.
|
||||
|
||||
@router.get("/dashboard")
|
||||
async def dashboard(
|
||||
user_service: UserService = Depends(get_user_service),
|
||||
item_service: ItemService = Depends(get_item_service),
|
||||
# Both services share the same db session from get_db()
|
||||
):
|
||||
users = await user_service.count()
|
||||
items = await item_service.count()
|
||||
return {"users": users, "items": items}
|
||||
|
||||
# To disable caching (get fresh instance each time):
|
||||
@router.get("/example")
|
||||
async def example(
|
||||
db1: AsyncSession = Depends(get_db),
|
||||
db2: AsyncSession = Depends(get_db, use_cache=False),
|
||||
# db1 and db2 are different sessions
|
||||
):
|
||||
pass
|
||||
```
|
||||
|
||||
#### Class-based dependencies
|
||||
|
||||
```python
|
||||
class Pagination:
|
||||
def __init__(
|
||||
self,
|
||||
skip: int = Query(0, ge=0),
|
||||
limit: int = Query(20, ge=1, le=100),
|
||||
):
|
||||
self.skip = skip
|
||||
self.limit = limit
|
||||
|
||||
@router.get("/items")
|
||||
async def list_items(pagination: Pagination = Depends()):
|
||||
# pagination.skip, pagination.limit
|
||||
pass
|
||||
```
|
||||
|
||||
### 4. Middleware
|
||||
|
||||
#### Custom middleware
|
||||
|
||||
```python
|
||||
import time
|
||||
from fastapi import Request
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
|
||||
class TimingMiddleware(BaseHTTPMiddleware):
|
||||
async def dispatch(self, request: Request, call_next):
|
||||
start = time.perf_counter()
|
||||
response = await call_next(request)
|
||||
duration = time.perf_counter() - start
|
||||
response.headers["X-Process-Time"] = f"{duration:.4f}"
|
||||
return response
|
||||
```
|
||||
|
||||
#### Pure ASGI middleware (higher performance)
|
||||
|
||||
```python
|
||||
from starlette.types import ASGIApp, Receive, Scope, Send
|
||||
|
||||
class RequestIDMiddleware:
|
||||
def __init__(self, app: ASGIApp):
|
||||
self.app = app
|
||||
|
||||
async def __call__(self, scope: Scope, receive: Receive, send: Send):
|
||||
if scope["type"] == "http":
|
||||
request_id = uuid.uuid4().hex
|
||||
scope.setdefault("state", {})["request_id"] = request_id
|
||||
|
||||
async def send_with_header(message):
|
||||
if message["type"] == "http.response.start":
|
||||
headers = dict(message.get("headers", []))
|
||||
headers[b"x-request-id"] = request_id.encode()
|
||||
message["headers"] = list(headers.items())
|
||||
await send(message)
|
||||
|
||||
await self.app(scope, receive, send_with_header)
|
||||
else:
|
||||
await self.app(scope, receive, send)
|
||||
```
|
||||
|
||||
#### Standard middleware configuration
|
||||
|
||||
```python
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.middleware.trustedhost import TrustedHostMiddleware
|
||||
from fastapi.middleware.gzip import GZipMiddleware
|
||||
|
||||
def add_middleware(app: FastAPI):
|
||||
# Order matters: first added = outermost (runs first on request, last on response)
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=settings.CORS_ORIGINS, # ["https://example.com"]
|
||||
allow_credentials=True,
|
||||
allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
app.add_middleware(
|
||||
TrustedHostMiddleware,
|
||||
allowed_hosts=settings.ALLOWED_HOSTS, # ["example.com", "*.example.com"]
|
||||
)
|
||||
|
||||
app.add_middleware(GZipMiddleware, minimum_size=500) # Compress responses > 500 bytes
|
||||
|
||||
app.add_middleware(TimingMiddleware)
|
||||
```
|
||||
|
||||
### 5. Background Tasks
|
||||
|
||||
#### Simple background tasks
|
||||
|
||||
```python
|
||||
from fastapi import BackgroundTasks
|
||||
|
||||
async def send_welcome_email(email: str, name: str):
|
||||
# This runs after the response is sent
|
||||
await email_service.send(
|
||||
to=email,
|
||||
subject="Welcome!",
|
||||
body=f"Hello {name}, welcome to our platform.",
|
||||
)
|
||||
|
||||
async def log_activity(user_id: int, action: str):
|
||||
await activity_repo.create(user_id=user_id, action=action)
|
||||
|
||||
@router.post("/users", status_code=201)
|
||||
async def create_user(
|
||||
user: UserCreate,
|
||||
background_tasks: BackgroundTasks,
|
||||
):
|
||||
new_user = await user_service.create(user)
|
||||
# Queue multiple background tasks
|
||||
background_tasks.add_task(send_welcome_email, new_user.email, new_user.name)
|
||||
background_tasks.add_task(log_activity, new_user.id, "account_created")
|
||||
return new_user
|
||||
```
|
||||
|
||||
#### Long-running tasks with task queues
|
||||
|
||||
For tasks that take more than a few seconds, use a proper task queue:
|
||||
|
||||
```python
|
||||
from celery import Celery
|
||||
|
||||
celery_app = Celery("worker", broker=settings.CELERY_BROKER_URL)
|
||||
|
||||
@celery_app.task
|
||||
def generate_report(report_id: int):
|
||||
# Long-running: query data, build PDF, upload to S3
|
||||
...
|
||||
|
||||
@router.post("/reports", status_code=202)
|
||||
async def request_report(params: ReportRequest):
|
||||
report = await report_service.create(params)
|
||||
generate_report.delay(report.id) # Dispatch to Celery worker
|
||||
return {"report_id": report.id, "status": "processing"}
|
||||
|
||||
@router.get("/reports/{report_id}/status")
|
||||
async def report_status(report_id: int):
|
||||
report = await report_service.get(report_id)
|
||||
return {"status": report.status, "url": report.download_url}
|
||||
```
|
||||
|
||||
### 6. WebSocket
|
||||
|
||||
#### WebSocket endpoint with connection management
|
||||
|
||||
```python
|
||||
from fastapi import WebSocket, WebSocketDisconnect
|
||||
|
||||
class ConnectionManager:
|
||||
def __init__(self):
|
||||
self.active_connections: dict[str, list[WebSocket]] = {}
|
||||
|
||||
async def connect(self, websocket: WebSocket, room: str):
|
||||
await websocket.accept()
|
||||
self.active_connections.setdefault(room, []).append(websocket)
|
||||
|
||||
def disconnect(self, websocket: WebSocket, room: str):
|
||||
self.active_connections.get(room, []).remove(websocket)
|
||||
|
||||
async def broadcast(self, message: str, room: str):
|
||||
for connection in self.active_connections.get(room, []):
|
||||
try:
|
||||
await connection.send_text(message)
|
||||
except WebSocketDisconnect:
|
||||
self.disconnect(connection, room)
|
||||
|
||||
manager = ConnectionManager()
|
||||
|
||||
@app.websocket("/ws/{room}")
|
||||
async def websocket_endpoint(websocket: WebSocket, room: str):
|
||||
await manager.connect(websocket, room)
|
||||
try:
|
||||
while True:
|
||||
data = await websocket.receive_text()
|
||||
await manager.broadcast(f"Message: {data}", room)
|
||||
except WebSocketDisconnect:
|
||||
manager.disconnect(websocket, room)
|
||||
await manager.broadcast(f"User left the room", room)
|
||||
```
|
||||
|
||||
#### WebSocket with authentication
|
||||
|
||||
```python
|
||||
@app.websocket("/ws/private")
|
||||
async def private_ws(websocket: WebSocket, token: str = Query(...)):
|
||||
try:
|
||||
user = verify_token(token)
|
||||
except InvalidToken:
|
||||
await websocket.close(code=4001, reason="Invalid token")
|
||||
return
|
||||
|
||||
await websocket.accept()
|
||||
try:
|
||||
while True:
|
||||
data = await websocket.receive_json()
|
||||
response = await process_message(user, data)
|
||||
await websocket.send_json(response)
|
||||
except WebSocketDisconnect:
|
||||
pass
|
||||
```
|
||||
|
||||
### 7. File Handling
|
||||
|
||||
#### Upload files
|
||||
|
||||
```python
|
||||
from fastapi import UploadFile, File
|
||||
|
||||
@router.post("/upload")
|
||||
async def upload_file(
|
||||
file: UploadFile = File(..., description="File to upload"),
|
||||
):
|
||||
# Validate file type and size
|
||||
if file.content_type not in ["image/png", "image/jpeg"]:
|
||||
raise HTTPException(400, "Only PNG and JPEG images are allowed")
|
||||
|
||||
if file.size and file.size > 5 * 1024 * 1024: # 5 MB
|
||||
raise HTTPException(400, "File too large (max 5 MB)")
|
||||
|
||||
contents = await file.read()
|
||||
path = f"uploads/{uuid.uuid4()}_{file.filename}"
|
||||
async with aiofiles.open(path, "wb") as f:
|
||||
await f.write(contents)
|
||||
|
||||
return {"filename": file.filename, "path": path, "size": len(contents)}
|
||||
|
||||
# Multiple file upload
|
||||
@router.post("/upload-multiple")
|
||||
async def upload_multiple(files: list[UploadFile] = File(...)):
|
||||
results = []
|
||||
for file in files:
|
||||
contents = await file.read()
|
||||
results.append({"filename": file.filename, "size": len(contents)})
|
||||
return results
|
||||
```
|
||||
|
||||
#### Streaming responses
|
||||
|
||||
```python
|
||||
from fastapi.responses import StreamingResponse
|
||||
import csv
|
||||
import io
|
||||
|
||||
@router.get("/export/users")
|
||||
async def export_users():
|
||||
async def generate_csv():
|
||||
output = io.StringIO()
|
||||
writer = csv.writer(output)
|
||||
writer.writerow(["id", "name", "email"])
|
||||
yield output.getvalue()
|
||||
output.seek(0)
|
||||
output.truncate(0)
|
||||
|
||||
async for user in user_service.stream_all():
|
||||
writer.writerow([user.id, user.name, user.email])
|
||||
yield output.getvalue()
|
||||
output.seek(0)
|
||||
output.truncate(0)
|
||||
|
||||
return StreamingResponse(
|
||||
generate_csv(),
|
||||
media_type="text/csv",
|
||||
headers={"Content-Disposition": "attachment; filename=users.csv"},
|
||||
)
|
||||
```
|
||||
|
||||
#### Static files
|
||||
|
||||
```python
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
app.mount("/static", StaticFiles(directory="static"), name="static")
|
||||
```
|
||||
|
||||
### 8. Testing
|
||||
|
||||
#### TestClient for synchronous tests
|
||||
|
||||
```python
|
||||
from fastapi.testclient import TestClient
|
||||
from app.main import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
def test_create_user():
|
||||
response = client.post("/api/v1/users", json={
|
||||
"email": "test@example.com",
|
||||
"name": "Test User",
|
||||
})
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["email"] == "test@example.com"
|
||||
|
||||
def test_get_user_not_found():
|
||||
response = client.get("/api/v1/users/99999")
|
||||
assert response.status_code == 404
|
||||
```
|
||||
|
||||
#### Async testing with httpx
|
||||
|
||||
```python
|
||||
import pytest
|
||||
from httpx import AsyncClient, ASGITransport
|
||||
from app.main import app
|
||||
|
||||
@pytest.fixture
|
||||
async def async_client():
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
yield client
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_list_users(async_client: AsyncClient):
|
||||
response = await async_client.get("/api/v1/users")
|
||||
assert response.status_code == 200
|
||||
assert isinstance(response.json(), list)
|
||||
```
|
||||
|
||||
#### Overriding dependencies for tests
|
||||
|
||||
```python
|
||||
from app.dependencies import get_db, get_current_user
|
||||
from app.models.user import User
|
||||
|
||||
# Mock database session
|
||||
async def override_get_db():
|
||||
async with test_session_maker() as session:
|
||||
yield session
|
||||
|
||||
# Mock authenticated user
|
||||
async def override_get_current_user():
|
||||
return User(id=1, email="test@example.com", name="Test")
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
app.dependency_overrides[get_current_user] = override_get_current_user
|
||||
|
||||
# In conftest.py — clean up after tests
|
||||
@pytest.fixture(autouse=True)
|
||||
def clear_overrides():
|
||||
yield
|
||||
app.dependency_overrides.clear()
|
||||
```
|
||||
|
||||
#### Testing WebSocket endpoints
|
||||
|
||||
```python
|
||||
def test_websocket():
|
||||
with client.websocket_connect("/ws/test-room") as ws:
|
||||
ws.send_text("hello")
|
||||
data = ws.receive_text()
|
||||
assert "hello" in data
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Best Practices
|
||||
|
||||
1. Use Pydantic models for request/response validation
|
||||
2. Organize routes with APIRouter
|
||||
3. Use dependency injection for services
|
||||
4. Return proper HTTP status codes
|
||||
5. Add OpenAPI descriptions
|
||||
1. **Use Pydantic models for all request/response validation** — never pass raw dicts through your API boundary. Define separate `Create`, `Update`, and `Response` schemas for each resource.
|
||||
|
||||
2. **Organize routes with APIRouter** — group related endpoints by resource and version. Apply shared dependencies at the router level, not on each individual route.
|
||||
|
||||
3. **Separate business logic from routes** — route functions should only handle HTTP concerns (parsing request, returning response). Delegate logic to service classes injected via `Depends()`.
|
||||
|
||||
4. **Use the lifespan context manager** — replace deprecated `on_event("startup")` and `on_event("shutdown")` with the `lifespan` async context manager for resource setup and teardown.
|
||||
|
||||
5. **Return proper HTTP status codes** — 201 for creation, 204 for deletion, 202 for accepted-but-not-done, 409 for conflicts. Use `status_code` parameter on route decorators.
|
||||
|
||||
6. **Add OpenAPI metadata** — provide `summary`, `description`, `tags`, and `responses` on routes. Set `title`, `version`, and `description` on the FastAPI app. This generates high-quality auto-docs.
|
||||
|
||||
7. **Use async all the way down** — if your route is `async def`, every I/O call inside it must also be async. Mixing sync blocking calls (e.g., `requests.get()`) in an async route will block the event loop.
|
||||
|
||||
8. **Configure settings with pydantic-settings** — load config from environment variables with validation and type coercion. Never hardcode secrets or connection strings.
|
||||
|
||||
```python
|
||||
from pydantic_settings import BaseSettings
|
||||
|
||||
class Settings(BaseSettings):
|
||||
DATABASE_URL: str
|
||||
API_KEY: str
|
||||
DEBUG: bool = False
|
||||
|
||||
model_config = {"env_file": ".env"}
|
||||
|
||||
settings = Settings()
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Common Pitfalls
|
||||
|
||||
- **Blocking I/O in async**: Use async libraries
|
||||
- **Missing response models**: Always define them
|
||||
- **No error handling**: Use HTTPException properly
|
||||
1. **Blocking I/O in async routes** — calling `requests.get()`, `time.sleep()`, or synchronous DB drivers inside `async def` routes starves the event loop. Use `httpx`, `asyncio.sleep()`, and async database drivers instead. If you must call sync code, use `run_in_executor`.
|
||||
|
||||
2. **Missing response_model** — without `response_model`, FastAPI returns whatever you return, potentially leaking internal fields (passwords, internal IDs). Always define a Pydantic response schema.
|
||||
|
||||
3. **Forgetting to await coroutines** — calling `await db.execute(query)` vs `db.execute(query)` is easy to miss. The latter returns a coroutine object instead of results. Enable linting rules that catch unawaited coroutines.
|
||||
|
||||
4. **Circular imports between models and schemas** — when schemas reference ORM models and vice versa, you get import cycles. Fix by using `TYPE_CHECKING` imports or by keeping schemas and models in separate modules that do not import each other.
|
||||
|
||||
5. **Not handling Pydantic validation errors** — FastAPI returns 422 by default, but the error format may confuse API consumers. Add a custom exception handler to reshape validation error responses to match your API's error format.
|
||||
|
||||
6. **Sharing mutable state across requests without locks** — global mutable variables (lists, dicts) accessed from async routes can cause race conditions. Use async-safe structures or dependency-injected per-request state.
|
||||
|
||||
---
|
||||
|
||||
## Related Skills
|
||||
|
||||
- `languages/python` — Python language patterns and best practices
|
||||
- `api/openapi` — OpenAPI specification and documentation standards
|
||||
- `databases/postgresql` — Database integration with async SQLAlchemy
|
||||
- `testing/pytest` — Testing FastAPI applications with pytest and httpx
|
||||
- `patterns/authentication` — JWT, OAuth2, and session patterns for FastAPI endpoints
|
||||
- `patterns/logging` — Structured logging for FastAPI applications
|
||||
|
||||
@@ -0,0 +1,229 @@
|
||||
# FastAPI Project Structure Reference
|
||||
|
||||
## Small Project (1-5 endpoints, single module)
|
||||
|
||||
```
|
||||
project/
|
||||
├── main.py # App factory, routes, startup
|
||||
├── models.py # Pydantic schemas + SQLAlchemy models
|
||||
├── database.py # DB connection, session factory
|
||||
├── config.py # Settings via pydantic-settings
|
||||
├── requirements.txt
|
||||
├── .env
|
||||
└── tests/
|
||||
├── conftest.py # Fixtures (test client, test DB)
|
||||
└── test_main.py
|
||||
```
|
||||
|
||||
**When to use**: Prototypes, microservices, internal tools, single-domain APIs.
|
||||
|
||||
**`main.py` structure**:
|
||||
```python
|
||||
from fastapi import FastAPI
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
# startup
|
||||
yield
|
||||
# shutdown
|
||||
|
||||
app = FastAPI(lifespan=lifespan)
|
||||
|
||||
@app.get("/health")
|
||||
async def health(): return {"status": "ok"}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Medium Project (5-20 endpoints, feature-grouped)
|
||||
|
||||
```
|
||||
project/
|
||||
├── app/
|
||||
│ ├── __init__.py
|
||||
│ ├── main.py # App factory, include routers
|
||||
│ ├── config.py # Settings (pydantic-settings)
|
||||
│ ├── database.py # Engine, SessionLocal, Base
|
||||
│ ├── dependencies.py # Shared deps (get_db, get_current_user)
|
||||
│ ├── exceptions.py # Custom exception handlers
|
||||
│ ├── middleware.py # CORS, logging, timing middleware
|
||||
│ │
|
||||
│ ├── auth/
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── router.py # POST /login, POST /register
|
||||
│ │ ├── schemas.py # LoginRequest, TokenResponse
|
||||
│ │ ├── models.py # User SQLAlchemy model
|
||||
│ │ ├── service.py # Business logic (hash, verify, tokens)
|
||||
│ │ └── dependencies.py # get_current_user, require_role
|
||||
│ │
|
||||
│ ├── items/
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── router.py # CRUD endpoints
|
||||
│ │ ├── schemas.py # ItemCreate, ItemRead, ItemUpdate
|
||||
│ │ ├── models.py # Item SQLAlchemy model
|
||||
│ │ └── service.py # Business logic
|
||||
│ │
|
||||
│ └── shared/
|
||||
│ ├── __init__.py
|
||||
│ ├── pagination.py # Pagination params + response schema
|
||||
│ └── filters.py # Common query filter patterns
|
||||
│
|
||||
├── alembic/ # DB migrations
|
||||
│ ├── env.py
|
||||
│ └── versions/
|
||||
├── alembic.ini
|
||||
├── requirements.txt
|
||||
├── pyproject.toml
|
||||
├── Dockerfile
|
||||
├── docker-compose.yml
|
||||
└── tests/
|
||||
├── conftest.py
|
||||
├── auth/
|
||||
│ └── test_router.py
|
||||
└── items/
|
||||
├── test_router.py
|
||||
└── test_service.py
|
||||
```
|
||||
|
||||
**When to use**: Multi-feature APIs, team projects, typical SaaS backends.
|
||||
|
||||
**Key patterns**:
|
||||
- Each feature gets its own directory with router, schemas, models, service
|
||||
- `router.py` uses `APIRouter(prefix="/items", tags=["items"])`
|
||||
- `main.py` includes routers: `app.include_router(items.router)`
|
||||
- Shared deps in root `dependencies.py`, feature-specific in feature dir
|
||||
|
||||
---
|
||||
|
||||
## Large Project (20+ endpoints, domain-driven)
|
||||
|
||||
```
|
||||
project/
|
||||
├── src/
|
||||
│ ├── __init__.py
|
||||
│ ├── main.py # App factory only
|
||||
│ ├── config.py # Layered settings
|
||||
│ │
|
||||
│ ├── core/ # Framework-level concerns
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── database.py # Engine, session management
|
||||
│ │ ├── security.py # JWT, hashing, RBAC
|
||||
│ │ ├── exceptions.py # Base exceptions + handlers
|
||||
│ │ ├── middleware.py # All middleware stack
|
||||
│ │ ├── dependencies.py # Cross-cutting deps
|
||||
│ │ ├── events.py # Domain event bus
|
||||
│ │ └── pagination.py # Cursor + offset pagination
|
||||
│ │
|
||||
│ ├── domain/ # Business logic (framework-agnostic)
|
||||
│ │ ├── users/
|
||||
│ │ │ ├── __init__.py
|
||||
│ │ │ ├── entity.py # Domain entity (plain dataclass)
|
||||
│ │ │ ├── repository.py # Abstract repository interface
|
||||
│ │ │ ├── service.py # Business rules
|
||||
│ │ │ └── events.py # Domain events
|
||||
│ │ ├── orders/
|
||||
│ │ │ └── ...
|
||||
│ │ └── payments/
|
||||
│ │ └── ...
|
||||
│ │
|
||||
│ ├── infrastructure/ # External system adapters
|
||||
│ │ ├── database/
|
||||
│ │ │ ├── models.py # All SQLAlchemy models
|
||||
│ │ │ ├── repositories/ # Concrete repo implementations
|
||||
│ │ │ │ ├── user_repo.py
|
||||
│ │ │ │ └── order_repo.py
|
||||
│ │ │ └── migrations/ # Alembic
|
||||
│ │ ├── cache/
|
||||
│ │ │ └── redis_client.py
|
||||
│ │ ├── email/
|
||||
│ │ │ └── smtp_service.py
|
||||
│ │ └── external/
|
||||
│ │ └── stripe_client.py
|
||||
│ │
|
||||
│ └── api/ # HTTP layer only
|
||||
│ ├── __init__.py
|
||||
│ ├── v1/
|
||||
│ │ ├── __init__.py # v1 router aggregator
|
||||
│ │ ├── users.py # Thin: parse request -> call service -> format response
|
||||
│ │ ├── orders.py
|
||||
│ │ └── payments.py
|
||||
│ ├── v2/
|
||||
│ │ └── ...
|
||||
│ ├── schemas/ # Request/response schemas
|
||||
│ │ ├── user_schemas.py
|
||||
│ │ ├── order_schemas.py
|
||||
│ │ └── common.py
|
||||
│ ├── dependencies.py # API-layer deps
|
||||
│ └── websockets/
|
||||
│ └── notifications.py
|
||||
│
|
||||
├── tests/
|
||||
│ ├── conftest.py
|
||||
│ ├── unit/
|
||||
│ │ ├── domain/
|
||||
│ │ │ └── test_user_service.py
|
||||
│ │ └── ...
|
||||
│ ├── integration/
|
||||
│ │ ├── test_user_api.py
|
||||
│ │ └── test_order_flow.py
|
||||
│ └── e2e/
|
||||
│ └── test_checkout.py
|
||||
│
|
||||
├── scripts/ # Dev/ops scripts
|
||||
│ ├── seed_db.py
|
||||
│ └── migrate.py
|
||||
├── pyproject.toml
|
||||
├── Dockerfile
|
||||
├── docker-compose.yml
|
||||
└── Makefile
|
||||
```
|
||||
|
||||
**When to use**: Complex domains, multiple teams, long-lived products.
|
||||
|
||||
**Key patterns**:
|
||||
- **Domain layer** has zero framework imports (testable in isolation)
|
||||
- **Infrastructure** implements domain interfaces (repository pattern)
|
||||
- **API layer** is thin: validation, auth, call service, return schema
|
||||
- API versioning via `/api/v1/`, `/api/v2/`
|
||||
- Separate unit, integration, and e2e test directories
|
||||
|
||||
---
|
||||
|
||||
## File Responsibilities
|
||||
|
||||
| File | Responsibility | Dependencies |
|
||||
|------|---------------|-------------|
|
||||
| `router.py` | HTTP handling, request parsing, response formatting | schemas, service, dependencies |
|
||||
| `schemas.py` | Pydantic models for request/response validation | None (or shared schemas) |
|
||||
| `models.py` | SQLAlchemy/ODM models (DB table mapping) | database |
|
||||
| `service.py` | Business logic, orchestration | repository/models, external services |
|
||||
| `dependencies.py` | FastAPI `Depends()` callables | database, config, auth |
|
||||
| `exceptions.py` | Custom exceptions + handlers | None |
|
||||
| `config.py` | `BaseSettings` with env loading | None |
|
||||
|
||||
## Router Registration Pattern
|
||||
|
||||
```python
|
||||
# app/main.py
|
||||
from fastapi import FastAPI
|
||||
from app.auth.router import router as auth_router
|
||||
from app.items.router import router as items_router
|
||||
|
||||
def create_app() -> FastAPI:
|
||||
app = FastAPI(title="My API")
|
||||
app.include_router(auth_router)
|
||||
app.include_router(items_router)
|
||||
return app
|
||||
|
||||
app = create_app()
|
||||
```
|
||||
|
||||
```python
|
||||
# app/items/router.py
|
||||
from fastapi import APIRouter, Depends
|
||||
router = APIRouter(prefix="/items", tags=["items"])
|
||||
|
||||
@router.get("/")
|
||||
async def list_items(db=Depends(get_db)): ...
|
||||
```
|
||||
Reference in New Issue
Block a user