Skip to main content الرئيسية المنشئون bobmatnyc mcp-skillset fastapi-modern-web-development
fastapi-modern-web-development 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
الانتقال إلى التثبيت سوق المهارات اكتشف واستكشف مهارات الذكاء الاصطناعي التي بناها المجتمع.
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
عرض تفاصيل Prompt يتجاوز الأمر المباشر Prompt المخصّص للمراجعة. افحص المصدر قبل تشغيله.
npx skills add https://github.com/bobmatnyc/mcp-skillset --skill fastapi-modern-web-developmentيبقى الأمر في سطر واحد. مرّر أفقيًا لمراجعته كاملًا قبل النسخ.
تفضّل نسخة محلية؟ نزّل الملفات المتاحة حاليًا لدى SkillsMP.
تحميل Zip جاري التحميل... 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"]
FastAPI Modern Web Development
Overview
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.
When to Use This Skill
Use this skill when:
Building RESTful APIs for ML/AI services
Creating high-performance async Python web services
Developing data-intensive applications requiring concurrent request handling
Implementing microservices with automatic API documentation
Building APIs that require strong type safety and validation
Designing endpoints for LLM integration and AI workflows
Core Principles
1. Async-First Architecture
Always prefer async/await for I/O-bound operations
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
import httpx
app = FastAPI()
@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()
@app.get("/external-data"
async def fetch_external ():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/data" )
return
@app.get("/bad-example"
async
def
bad_handler
5
return
"status"
"done"
Why : FastAPI runs on ASGI (asyncio). Blocking calls prevent other requests from processing, degrading performance under load.
2. Pydantic v2 Models for Type Safety 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 )
id : int
username: str
email: str
created_at: datetime
@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())
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_validator
3. Dependency Injection for Loose Coupling Use 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
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
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
@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.
4. Structured Error Handling 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)
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
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" }
)
5. ML/AI Endpoint Design Patterns Optimize endpoints for ML model serving
from fastapi import BackgroundTasks
from functools import lru_cache
import asyncio
@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()
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None ,
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
)
@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" }
Best Practices
Application Structure 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
Configuration Management 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_URL: str = "postgresql+asyncpg://localhost/dbname"
SECRET_KEY: str
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
API_V1_PREFIX: str = "/api/v1"
PROJECT_NAME: str = "My FastAPI Project"
BACKEND_CORS_ORIGINS: list [str ] = ["http://localhost:3000" ]
@lru_cache()
def get_settings () -> Settings:
return Settings()
@app.get("/info"
async def info (settings: Settings = Depends(get_settings ) ):
return {"project_name" : settings.PROJECT_NAME}
Request Lifecycle & Middleware from fastapi import Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
import time
import logging
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000" ],
allow_credentials=True ,
allow_methods=["*" ],
allow_headers=["*" ],
)
app.add_middleware(GZipMiddleware, minimum_size=1000 )
@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:.3 f} s "
f"with status {response.status_code} "
)
response.headers["X-Process-Time" ] = str (process_time)
return response
Common Patterns
Pagination 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)
File Upload with Validation from fastapi import File, UploadFile
from PIL import Image
import aiofiles
@app.post("/upload-image"
async def upload_image (file: UploadFile = File(... ) ):
if not file.content_type.startswith("image/" ):
raise HTTPException(400 , "File must be an image" )
contents = await file.read()
if len (contents) > 10 * 1024 * 1024 :
raise HTTPException(400 , "File too large (max 10MB)" )
try :
image = Image.open (BytesIO(contents))
image.verify()
except Exception:
raise HTTPException(400 , "Invalid image file" )
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)}
WebSocket for Real-Time Updates 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)
Anti-Patterns to Avoid
❌ DON'T: Mix sync and async incorrectly
@app.get("/bad"
async def bad_endpoint ():
result = sync_database_call()
return result
@app.get("/good"
async def good_endpoint ():
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None , sync_database_call)
return result
❌ DON'T: Return ORM models directly
@app.get("/users/{user_id}"
async def get_user (user_id: int , db: AsyncSession = Depends(get_db ) ):
return await db.get(User, user_id)
@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
❌ DON'T: Hardcode configuration
DATABASE_URL = "postgresql://localhost/db"
settings = get_settings()
DATABASE_URL = settings.DATABASE_URL
❌ DON'T: Ignore startup/shutdown events
@app.on_event("startup"
async def startup ():
app.state.db = await create_db_pool()
@app.on_event("shutdown"
async def shutdown ():
await app.state.db.close()
Testing Strategy from fastapi.testclient import TestClient
from httpx import AsyncClient
import pytest
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"
@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
def override_get_db ():
return test_database_session
app.dependency_overrides[get_db] = override_get_db
Performance Optimization
Connection Pooling : Use SQLAlchemy async engine with pool_size=20, max_overflow=0Response Caching : Use Redis with @lru_cache() for expensive computationsBackground Tasks : Offload non-critical work to BackgroundTasksAsync Libraries : Use httpx (not requests), asyncpg (not psycopg2)Profiling : Use uvicorn --log-level debug and timing middleware
Security Checklist
✅ Use HTTPS in production (via reverse proxy)
✅ Implement rate limiting (slowapi library)
✅ Validate all inputs with Pydantic models
✅ Use OAuth2 with JWT for authentication
✅ Never expose internal error details in production
✅ Set proper CORS origins (not ["*"] in production)
✅ Hash passwords with bcrypt or argon2
✅ Use SECRET_KEY from environment, never hardcode
Related Skills
test-driven-development : Write tests before implementing endpointssystematic-debugging : Debug async issues and race conditionspostgresql-optimization : Optimize database queries for FastAPIsecurity-testing : Perform security audits on API endpoints
Additional Resources
Example Questions to Ask
"How do I implement JWT authentication in FastAPI?"
"What's the best way to handle file uploads asynchronously?"
"How do I optimize this endpoint for ML model inference?"
"Show me how to add pagination to this query"
"How do I write async tests for FastAPI endpoints?"
"What's the correct way to handle database transactions?"
المهن ذات الصلة SOC
استنادا إلى تصنيف SOC المهني
