メニュー
Production-grade FastAPI development with async patterns, Pydantic v2, dependency injection, ML/AI endpoint design, and modern Python best practices for building high-performance REST APIs
デフォルトでは、最初にソースを確認する Prompt が選択されています。直接コマンドに切り替えるか、ローカルコピーをダウンロードすることもできます。
インストールを決める前に、SKILL.md と SkillsMP に表示されている付属ファイルをお読みください。
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
直接コマンドでは確認用 Prompt が省略されます。実行前にソースを確認してください。
npx skills add https://github.com/bobmatnyc/mcp-skillset --skill fastapi-modern-web-developmentコマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか？SkillsMP が現在取得できるファイルをダウンロードできます。
Build ultra-low-latency edge computing applications with Cloudflare Workers, Workers AI for LLM inference, V8 isolates, Durable Objects, and serverless patterns deployed across 330+ data centers worldwide
Production-grade observability stack with Prometheus metrics, Grafana dashboards, PromQL query language, alerting rules, and AI-powered anomaly detection for modern cloud-native applications
Production-grade PostgreSQL query optimization, indexing strategies, performance tuning, and modern features including pgvector for AI/ML workloads. Master EXPLAIN plans, query analysis, and database design for high-performance applications
SOC 職業分類に基づく
SKILL.md を表示中
| name | FastAPI Modern Web Development |
| skill_id | fastapi-web-development |
| version | 1.0.0 |
| description | Production-grade FastAPI development with async patterns, Pydantic v2, dependency injection, ML/AI endpoint design, and modern Python best practices for building high-performance REST APIs |
| category | Python Web Development |
| tags | ["fastapi","python","async","pydantic","rest-api","ml-endpoints","dependency-injection","web-framework","async-programming","api-design"] |
| author | mcp-skillset |
| license | MIT |
| created | 2025-11-25T00:00:00.000Z |
| last_updated | 2025-11-25T00:00:00.000Z |
| toolchain | ["Python 3.11+","FastAPI 0.100+","Pydantic v2","uvicorn","httpx"] |
| frameworks | ["FastAPI","Pydantic","SQLAlchemy 2.0","Alembic"] |
| related_skills | ["test-driven-development","systematic-debugging","postgresql-optimization","security-testing"] |
This skill provides comprehensive guidance for building production-grade FastAPI applications with modern Python patterns (2024-2025 best practices). FastAPI is the #1 framework for AI/ML APIs, combining high performance, automatic OpenAPI documentation, and intuitive async/await patterns.
Use this skill when:
Always prefer async/await for I/O-bound operations
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
import httpx
app = FastAPI()
# CORRECT: Async database operations
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User).where(User.id == user_id))
return result.scalar_one_or_none()
# CORRECT: Async external API calls
@app.get("/external-data")
async def fetch_external():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/data")
response.json()
():
time.sleep()
{: }
Why: FastAPI runs on ASGI (asyncio). Blocking calls prevent other requests from processing, degrading performance under load.
Use Pydantic models for all request/response validation
from pydantic import BaseModel, Field, field_validator, ConfigDict
from datetime import datetime
from typing import Optional
class UserCreate(BaseModel):
"""Request model for user creation"""
model_config = ConfigDict(str_strip_whitespace=True)
username: str = Field(..., min_length=3, max_length=50)
email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
age: Optional[int] = Field(None, ge=13, le=120)
@field_validator('username')
@classmethod
def username_alphanumeric(cls, v: str) -> str:
if not v.isalnum():
raise ValueError('Username must be alphanumeric')
return v.lower()
class UserResponse(BaseModel):
"""Response model - never expose internal fields"""
model_config = ConfigDict(from_attributes=True) # Pydantic v2
id: int
username: str
email: str
created_at: datetime
# DON'T expose: password_hash, internal_flags, etc.
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):
db_user = User(**user.model_dump()) # Pydantic v2 syntax
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
Key Updates for Pydantic v2:
Config → model_config = ConfigDict(...)orm_mode=True → from_attributes=Truedict() → model_dump()@validator → @field_validatorUse FastAPI's dependency injection for reusable components
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
# Database session dependency
async_engine = create_async_engine("postgresql+asyncpg://...")
AsyncSessionLocal = sessionmaker(
async_engine, class_=AsyncSession, expire_on_commit=False
)
async def get_db() -> AsyncSession:
async with AsyncSessionLocal() as session:
yield session
# Authentication dependency
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
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"},
)
payload = verify_token(token)
if payload is None:
raise credentials_exception
user = await db.get(User, payload.get("sub"))
if user is None:
raise credentials_exception
return user
# Use dependencies in route
@app.get("/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
Benefits: Automatic injection, easy testing with override_dependency, shared logic across routes.
Always use HTTPException with proper status codes
from fastapi import HTTPException, status
@app.get("/items/{item_id}")
async def get_item(item_id: int, db: AsyncSession = Depends(get_db)):
item = await db.get(Item, item_id)
# CORRECT: Specific HTTP exception
if item is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Item {item_id} not found"
)
if not item.is_public and not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Insufficient permissions"
)
return item
# Custom exception handler for domain errors
class ItemNotFoundError(Exception):
def __init__(self, item_id: int):
self.item_id = item_id
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request, exc: ItemNotFoundError):
return JSONResponse(
status_code=404,
content={"message": f"Item {exc.item_id} not found"}
)
Optimize endpoints for ML model serving
from fastapi import BackgroundTasks
from functools import lru_cache
import asyncio
# Singleton pattern for model loading
@lru_cache()
def get_ml_model():
"""Load model once, cache for all requests"""
import torch
model = torch.load("model.pth")
model.eval()
return model
class PredictionRequest(BaseModel):
text: str = Field(..., max_length=10000)
temperature: float = Field(0.7, ge=0.0, le=2.0)
class PredictionResponse(BaseModel):
prediction: str
confidence: float
processing_time_ms: float
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
model = get_ml_model()
start_time = asyncio.get_event_loop().time()
# Run CPU-intensive model inference in thread pool
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None, # Uses default ThreadPoolExecutor
model.predict,
request.text
)
elapsed = (asyncio.get_event_loop().time() - start_time) * 1000
return PredictionResponse(
prediction=result["text"],
confidence=result["score"],
processing_time_ms=elapsed
)
# Background task for async processing
@app.post("/predict-async")
async def predict_async(
request: PredictionRequest,
background_tasks: BackgroundTasks
):
task_id = generate_task_id()
background_tasks.add_task(process_prediction, task_id, request)
return {"task_id": task_id, "status": "processing"}
fastapi-project/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI app, startup/shutdown events
│ ├── config.py # Pydantic Settings management
│ ├── models/ # SQLAlchemy ORM models
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ ├── schemas/ # Pydantic models (request/response)
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── item.py
│ ├── api/ # API routes
│ │ ├── __init__.py
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ ├── users.py
│ │ │ └── items.py
│ │ └── deps.py # Shared dependencies
│ ├── services/ # Business logic
│ │ ├── __init__.py
│ │ ├── user_service.py
│ │ └── ml_service.py
│ ├── core/ # Core utilities
│ │ ├── __init__.py
│ │ ├── security.py
│ │ └── database.py
│ └── tests/
│ ├── __init__.py
│ ├── conftest.py
│ └── test_users.py
├── alembic/ # Database migrations
├── .env
├── pyproject.toml
└── README.md
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False
)
# Database
DATABASE_URL: str = "postgresql+asyncpg://localhost/dbname"
# Security
SECRET_KEY: str
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# API
API_V1_PREFIX: str = "/api/v1"
PROJECT_NAME: str = "My FastAPI Project"
# CORS
BACKEND_CORS_ORIGINS: list[str] = ["http://localhost:3000"]
@lru_cache()
def get_settings() -> Settings:
return Settings()
# Use in dependencies
@app.get("/info")
async def info(settings: Settings = Depends(get_settings)):
return {"project_name": settings.PROJECT_NAME}
from fastapi import Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time
import logging
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Compression middleware
app.add_middleware(GZipMiddleware, minimum_size=1000)
# Custom logging middleware
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
logging.info(
f"{request.method} {request.url.path} "
f"completed in {process_time:.3f}s "
f"with status {response.status_code}"
)
response.headers["X-Process-Time"] = str(process_time)
return response
from typing import Generic, TypeVar, Sequence
from pydantic import BaseModel
T = TypeVar('T')
class Page(BaseModel, Generic[T]):
items: Sequence[T]
total: int
page: int
size: int
pages: int
async def paginate(
query,
page: int = 1,
size: int = 50
) -> Page[T]:
total = await db.scalar(select(func.count()).select_from(query))
items = await db.scalars(
query.offset((page - 1) * size).limit(size)
)
return Page(
items=items.all(),
total=total,
page=page,
size=size,
pages=(total + size - 1) // size
)
@app.get("/users", response_model=Page[UserResponse])
async def list_users(
page: int = 1,
size: int = 50,
db: AsyncSession = Depends(get_db)
):
query = select(User).order_by(User.created_at.desc())
return await paginate(query, page, size)
from fastapi import File, UploadFile
from PIL import Image
import aiofiles
@app.post("/upload-image")
async def upload_image(file: UploadFile = File(...)):
# Validate file type
if not file.content_type.startswith("image/"):
raise HTTPException(400, "File must be an image")
# Validate file size (10MB limit)
contents = await file.read()
if len(contents) > 10 * 1024 * 1024:
raise HTTPException(400, "File too large (max 10MB)")
# Validate image can be opened
try:
image = Image.open(BytesIO(contents))
image.verify()
except Exception:
raise HTTPException(400, "Invalid image file")
# Save file asynchronously
file_path = f"uploads/{file.filename}"
async with aiofiles.open(file_path, 'wb') as f:
await f.write(contents)
return {"filename": file.filename, "size": len(contents)}
from fastapi import WebSocket, WebSocketDisconnect
class ConnectionManager:
def __init__(self):
self.active_connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Client says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
# WRONG: Sync function in async route
@app.get("/bad")
async def bad_endpoint():
result = sync_database_call() # Blocks event loop!
return result
# CORRECT: Use run_in_executor for sync calls
@app.get("/good")
async def good_endpoint():
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, sync_database_call)
return result
# WRONG: Exposes internal fields, risks lazy loading
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
return await db.get(User, user_id) # Returns ORM model!
# CORRECT: Use Pydantic response model
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
user = await db.get(User, user_id)
return user # FastAPI converts to UserResponse
# WRONG: Hardcoded values
DATABASE_URL = "postgresql://localhost/db"
# CORRECT: Use Pydantic Settings
settings = get_settings()
DATABASE_URL = settings.DATABASE_URL
# CORRECT: Cleanup resources
@app.on_event("startup")
async def startup():
# Initialize database pool, load ML models, etc.
app.state.db = await create_db_pool()
@app.on_event("shutdown")
async def shutdown():
# Close connections, cleanup
await app.state.db.close()
from fastapi.testclient import TestClient
from httpx import AsyncClient
import pytest
# Sync tests with TestClient
def test_create_user():
client = TestClient(app)
response = client.post(
"/users",
json={"username": "testuser", "email": "test@example.com"}
)
assert response.status_code == 201
assert response.json()["username"] == "testuser"
# Async tests with httpx
@pytest.mark.asyncio
async def test_async_endpoint():
async with AsyncClient(app=app, base_url="http://test") as client:
response = await client.get("/users/1")
assert response.status_code == 200
# Override dependencies for testing
def override_get_db():
return test_database_session
app.dependency_overrides[get_db] = override_get_db
@lru_cache() for expensive computationsuvicorn --log-level debug and timing middleware["*"] in production)SECRET_KEY from environment, never hardcode