python
Use when building FastAPI applications, implementing async endpoints, setting up Pydantic schemas, working with SQLAlchemy, or writing pytest tests for Python backend services.
Informations de source
- Dépôt
- MadAppGang/claude-code
- Dernière activité de la source
- 20 janvier 2026 à 08:16
- Langue détectée de SKILL.md
- anglais
- Étoiles
- 284
- Forks
- 25
Options d'installation
Le prompt qui vérifie d'abord la source est sélectionné par défaut. Vous pouvez passer à une commande directe ou télécharger une copie locale.
Vérifiez les fichiers source
Lisez SKILL.md et les fichiers associés affichés par SkillsMP avant de décider de l'installer.
Affichage de SKILL.md
SKILL.md
Instructions source · Aperçu en lecture seule- 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 Backend Patterns
## Overview
Python patterns for building backend services with FastAPI.
## Project Structure
```
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
```
## FastAPI Application
### Main Application
```python
# 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"}
```
### Configuration
```python
# 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()
```
## Pydantic Schemas
```python
# 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
```
## SQLAlchemy Models
```python
# 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())
```
## Async Database
```python
# 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()
```
## Repository Pattern
```python
# 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)
```
## API Routes
```python
# 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")
```
## Authentication
```python
# 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
```
## Testing
```python
# 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*
Voir sur GitHub