Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

10 NODES
🤖system prompt•7 months ago

fastapi-templates

Create production-ready FastAPI projects with async patterns,

coding
⭐1
# FastAPI Project Templates Production-ready FastAPI project structures with async patterns, dependency injection, middleware, and best practices for building high-performance APIs. ## When to Use This Skill - Starting new FastAPI projects from scratch - Implementing async REST APIs with Python - Building high-performance web services and microservices - Creating async applications with PostgreSQL, MongoDB - Setting up API projects with proper structure and testing ## Core Concepts ### 1. Project Structure **Recommended Layout:** ``` app/ ├── api/ # API routes │ ├── v1/ │ │ ├── endpoints/ │ │ │ ├── users.py │ │ │ ├── auth.py │ │ │ └── items.py │ │ └── router.py │ └── dependencies.py # Shared dependencies ├── core/ # Core configuration │ ├── config.py │ ├── security.py │ └── database.py ├── models/ # Database models │ ├── user.py │ └── item.py ├── schemas/ # Pydantic schemas │ ├── user.py │ └── item.py ├── services/ # Business logic │ ├── user_service.py │ └── auth_service.py ├── repositories/ # Data access │ ├── user_repository.py │ └── item_repository.py └── main.py # Application entry ``` ### 2. Dependency Injection FastAPI's built-in DI system using `Depends`: - Database session management - Authentication/authorization - Shared business logic - Configuration injection ### 3. Async Patterns Proper async/await usage: - Async route handlers - Async database operations - Async background tasks - Async middleware ## Implementation Patterns ### Pattern 1: Complete FastAPI Application ```python # main.py from fastapi import FastAPI, Depends from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan events.""" # Startup await database.connect() yield # Shutdown await database.disconnect() app = FastAPI( title="API Template", version="1.0.0", lifespan=lifespan ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Include routers from app.api.v1.router import api_router app.include_router(api_router, prefix="/api/v1") # core/config.py from pydantic_settings import BaseSettings from functools import lru_cache class Settings(BaseSettings): """Application settings.""" DATABASE_URL: str SECRET_KEY: str ACCESS_TOKEN_EXPIRE_MINUTES: int = 30 API_V1_STR: str = "/api/v1" class Config: env_file = ".env" @lru_cache() def get_settings() -> Settings: return Settings() # core/database.py from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from app.core.config import get_settings settings = get_settings() engine = create_async_engine( settings.DATABASE_URL, echo=True, future=True ) AsyncSessionLocal = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) Base = declarative_base() async def get_db() -> AsyncSession: """Dependency for database session.""" async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close() ``` ### Pattern 2: CRUD Repository Pattern ```python # repositories/base_repository.py from typing import Generic, TypeVar, Type, Optional, List from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from pydantic import BaseModel ModelType = TypeVar("ModelType") CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel) UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel) class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): """Base repository for CRUD operations.""" def __init__(self, model: Type[ModelType]): self.model = model async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]: """Get by ID.""" result = await db.execute( select(self.model).where(self.model.id == id) ) return result.scalars().first() async def get_multi( self, db: AsyncSession, skip: int = 0, limit: int = 100 ) -> List[ModelType]: """Get multiple records.""" result = await db.execute( select(self.model).offset(skip).limit(limit) ) return result.scalars().all() async def create( self, db: AsyncSession, obj_in: CreateSchemaType ) -> ModelType: """Create new record.""" db_obj = self.model(**obj_in.dict()) db.add(db_obj) await db.flush() await db.refresh(db_obj) return db_obj async def update( self, db: AsyncSession, db_obj: ModelType, obj_in: UpdateSchemaType ) -> ModelType: """Update record.""" update_data = obj_in.dict(exclude_unset=True) for field, value in update_data.items(): setattr(db_obj, field, value) await db.flush() await db.refresh(db_obj) return db_obj async def delete(self, db: AsyncSession, id: int) -> bool: """Delete record.""" obj = await self.get(db, id) if obj: await db.delete(obj) return True return False # repositories/user_repository.py from app.repositories.base_repository import BaseRepository from app.models.user import User from app.schemas.user import UserCreate, UserUpdate class UserRepository(BaseRepository[User, UserCreate, UserUpdate]): """User-specific repository.""" async def get_by_email(self, db: AsyncSession, email: str) -> Optional[User]: """Get user by email.""" result = await db.execute( select(User).where(User.email == email) ) return result.scalars().first() async def is_active(self, db: AsyncSession, user_id: int) -> bool: """Check if user is active.""" user = await self.get(db, user_id) return user.is_active if user else False user_repository = UserRepository(User) ``` ### Pattern 3: Service Layer ```python # services/user_service.py from typing import Optional from sqlalchemy.ext.asyncio import AsyncSession from app.repositories.user_repository import user_repository from app.schemas.user import UserCreate, UserUpdate, User from app.core.security import get_password_hash, verify_password class UserService: """Business logic for users.""" def __init__(self): self.repository = user_repository async def create_user( self, db: AsyncSession, user_in: UserCreate ) -> User: """Create new user with hashed password.""" # Check if email exists existing = await self.repository.get_by_email(db, user_in.email) if existing: raise ValueError("Email already registered") # Hash password user_in_dict = user_in.dict() user_in_dict["hashed_password"] = get_password_hash(user_in_dict.pop("password")) # Create user user = await self.repository.create(db, UserCreate(**user_in_dict)) return user async def authenticate( self, db: AsyncSession, email: str, password: str ) -> Optional[User]: """Authenticate user.""" user = await self.repository.get_by_email(db, email) if not user: return None if not verify_password(password, user.hashed_password): return None return user async def update_user( self, db: AsyncSession, user_id: int, user_in: UserUpdate ) -> Optional[User]: """Update user.""" user = await self.repository.get(db, user_id) if not user: return None if user_in.password: user_in_dict = user_in.dict(exclude_unset=True) user_in_dict["hashed_password"] = get_password_hash( user_in_dict.pop("password") ) user_in = UserUpdate(**user_in_dict) return await self.repository.update(db, user, user_in) user_service = UserService() ``` ### Pattern 4: API Endpoints with Dependencies ```python # api/v1/endpoints/users.py from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from typing import List from app.core.database import get_db from app.schemas.user import User, UserCreate, UserUpdate from app.services.user_service import user_service from app.api.dependencies import get_current_user router = APIRouter() @router.post("/", response_model=User, status_code=status.HTTP_201_CREATED) async def create_user( user_in: UserCreate, db: AsyncSession = Depends(get_db) ): """Create new user.""" try: user = await user_service.create_user(db, user_in) return user except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) @router.get("/me", response_model=User) async def read_current_user( current_user: User = Depends(get_current_user) ): """Get current user.""" return current_user @router.get("/{user_id}", response_model=User) async def read_user( user_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """Get user by ID.""" user = await user_service.repository.get(db, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user @router.patch("/{user_id}", response_model=User) async def update_user( user_id: int, user_in: UserUpdate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """Update user.""" if current_user.id != user_id: raise HTTPException(status_code=403, detail="Not authorized") user = await user_service.update_user(db, user_id, user_in) if not user: raise HTTPException(status_code=404, detail="User not found") return user @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_user( user_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """Delete user.""" if current_user.id != user_id: raise HTTPException(status_code=403, detail="Not authorized") deleted = await user_service.repository.delete(db, user_id) if not deleted: raise HTTPException(status_code=404, detail="User not found") ``` ### Pattern 5: Authentication & Authorization ```python # core/security.py from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from app.core.config import get_settings settings = get_settings() pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") ALGORITHM = "HS256" def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): """Create JWT access token.""" to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify password against hash.""" return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: """Hash password.""" return pwd_context.hash(password) # api/dependencies.py from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.security import ALGORITHM from app.core.config import get_settings from app.repositories.user_repository import user_repository oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login") async def get_current_user( db: AsyncSession = Depends(get_db), token: str = Depends(oauth2_scheme) ): """Get current authenticated 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=[ALGORITHM]) user_id: int = payload.get("sub") if user_id is None: raise credentials_exception except JWTError: raise credentials_exception user = await user_repository.get(db, user_id) if user is None: raise credentials_exception return user ``` ## Testing ```python # tests/conftest.py import pytest import asyncio from httpx import AsyncClient from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from app.main import app from app.core.database import get_db, Base TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:" @pytest.fixture(scope="session") def event_loop(): loop = asyncio.get_event_loop_policy().new_event_loop() yield loop loop.close() @pytest.fixture async def db_session(): engine = create_async_engine(TEST_DATABASE_URL, echo=True) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) AsyncSessionLocal = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) async with AsyncSessionLocal() as session: yield session @pytest.fixture async def client(db_session): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db async with AsyncClient(app=app, base_url="http://test") as client: yield client # tests/test_users.py import pytest @pytest.mark.asyncio async def test_create_user(client): response = await client.post( "/api/v1/users/", json={ "email": "test@example.com", "password": "testpass123", "name": "Test User" } ) assert response.status_code == 201 data = response.json() assert data["email"] == "test@example.com" assert "id" in data ``` ## Resources - **references/fastapi-architecture.md**: Detailed architecture guide - **references/async-best-practices.md**: Async/await patterns - **references/testing-strategies.md**: Comprehensive testing guide - **assets/project-template/**: Complete FastAPI project - **assets/docker-compose.yml**: Development environment setup ## Best Practices 1. **Async All The Way**: Use async for database, external APIs 2. **Dependency Injection**: Leverage FastAPI's DI system 3. **Repository Pattern**: Separate data access from business logic 4. **Service Layer**: Keep business logic out of routes 5. **Pydantic Schemas**: Strong typing for request/response 6. **Error Handling**: Consistent error responses 7. **Testing**: Test all layers independently ## Common Pitfalls - **Blocking Code in Async**: Using synchronous database drivers - **No Service Layer**: Business logic in route handlers - **Missing Type Hints**: Loses FastAPI's benefits - **Ignoring Sessions**: Not properly managing database sessions - **No Testing**: Skipping integration tests - **Tight Coupling**: Direct database access in routes
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

api-design-principles

Master REST and GraphQL API design principles to build intuitive,

coding
⭐1
# API Design Principles Master REST and GraphQL API design principles to build intuitive, scalable, and maintainable APIs that delight developers and stand the test of time. ## When to Use This Skill - Designing new REST or GraphQL APIs - Refactoring existing APIs for better usability - Establishing API design standards for your team - Reviewing API specifications before implementation - Migrating between API paradigms (REST to GraphQL, etc.) - Creating developer-friendly API documentation - Optimizing APIs for specific use cases (mobile, third-party integrations) ## Core Concepts ### 1. RESTful Design Principles **Resource-Oriented Architecture** - Resources are nouns (users, orders, products), not verbs - Use HTTP methods for actions (GET, POST, PUT, PATCH, DELETE) - URLs represent resource hierarchies - Consistent naming conventions **HTTP Methods Semantics:** - `GET`: Retrieve resources (idempotent, safe) - `POST`: Create new resources - `PUT`: Replace entire resource (idempotent) - `PATCH`: Partial resource updates - `DELETE`: Remove resources (idempotent) ### 2. GraphQL Design Principles **Schema-First Development** - Types define your domain model - Queries for reading data - Mutations for modifying data - Subscriptions for real-time updates **Query Structure:** - Clients request exactly what they need - Single endpoint, multiple operations - Strongly typed schema - Introspection built-in ### 3. API Versioning Strategies **URL Versioning:** ``` /api/v1/users /api/v2/users ``` **Header Versioning:** ``` Accept: application/vnd.api+json; version=1 ``` **Query Parameter Versioning:** ``` /api/users?version=1 ``` ## REST API Design Patterns ### Pattern 1: Resource Collection Design ```python # Good: Resource-oriented endpoints GET /api/users # List users (with pagination) POST /api/users # Create user GET /api/users/{id} # Get specific user PUT /api/users/{id} # Replace user PATCH /api/users/{id} # Update user fields DELETE /api/users/{id} # Delete user # Nested resources GET /api/users/{id}/orders # Get user's orders POST /api/users/{id}/orders # Create order for user # Bad: Action-oriented endpoints (avoid) POST /api/createUser POST /api/getUserById POST /api/deleteUser ``` ### Pattern 2: Pagination and Filtering ```python from typing import List, Optional from pydantic import BaseModel, Field class PaginationParams(BaseModel): page: int = Field(1, ge=1, description="Page number") page_size: int = Field(20, ge=1, le=100, description="Items per page") class FilterParams(BaseModel): status: Optional[str] = None created_after: Optional[str] = None search: Optional[str] = None class PaginatedResponse(BaseModel): items: List[dict] total: int page: int page_size: int pages: int @property def has_next(self) -> bool: return self.page < self.pages @property def has_prev(self) -> bool: return self.page > 1 # FastAPI endpoint example from fastapi import FastAPI, Query, Depends app = FastAPI() @app.get("/api/users", response_model=PaginatedResponse) async def list_users( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), status: Optional[str] = Query(None), search: Optional[str] = Query(None) ): # Apply filters query = build_query(status=status, search=search) # Count total total = await count_users(query) # Fetch page offset = (page - 1) * page_size users = await fetch_users(query, limit=page_size, offset=offset) return PaginatedResponse( items=users, total=total, page=page, page_size=page_size, pages=(total + page_size - 1) // page_size ) ``` ### Pattern 3: Error Handling and Status Codes ```python from fastapi import HTTPException, status from pydantic import BaseModel class ErrorResponse(BaseModel): error: str message: str details: Optional[dict] = None timestamp: str path: str class ValidationErrorDetail(BaseModel): field: str message: str value: Any # Consistent error responses STATUS_CODES = { "success": 200, "created": 201, "no_content": 204, "bad_request": 400, "unauthorized": 401, "forbidden": 403, "not_found": 404, "conflict": 409, "unprocessable": 422, "internal_error": 500 } def raise_not_found(resource: str, id: str): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "error": "NotFound", "message": f"{resource} not found", "details": {"id": id} } ) def raise_validation_error(errors: List[ValidationErrorDetail]): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={ "error": "ValidationError", "message": "Request validation failed", "details": {"errors": [e.dict() for e in errors]} } ) # Example usage @app.get("/api/users/{user_id}") async def get_user(user_id: str): user = await fetch_user(user_id) if not user: raise_not_found("User", user_id) return user ``` ### Pattern 4: HATEOAS (Hypermedia as the Engine of Application State) ```python class UserResponse(BaseModel): id: str name: str email: str _links: dict @classmethod def from_user(cls, user: User, base_url: str): return cls( id=user.id, name=user.name, email=user.email, _links={ "self": {"href": f"{base_url}/api/users/{user.id}"}, "orders": {"href": f"{base_url}/api/users/{user.id}/orders"}, "update": { "href": f"{base_url}/api/users/{user.id}", "method": "PATCH" }, "delete": { "href": f"{base_url}/api/users/{user.id}", "method": "DELETE" } } ) ``` ## GraphQL Design Patterns ### Pattern 1: Schema Design ```graphql # schema.graphql # Clear type definitions type User { id: ID! email: String! name: String! createdAt: DateTime! # Relationships orders(first: Int = 20, after: String, status: OrderStatus): OrderConnection! profile: UserProfile } type Order { id: ID! status: OrderStatus! total: Money! items: [OrderItem!]! createdAt: DateTime! # Back-reference user: User! } # Pagination pattern (Relay-style) type OrderConnection { edges: [OrderEdge!]! pageInfo: PageInfo! totalCount: Int! } type OrderEdge { node: Order! cursor: String! } type PageInfo { hasNextPage: Boolean! hasPreviousPage: Boolean! startCursor: String endCursor: String } # Enums for type safety enum OrderStatus { PENDING CONFIRMED SHIPPED DELIVERED CANCELLED } # Custom scalars scalar DateTime scalar Money # Query root type Query { user(id: ID!): User users(first: Int = 20, after: String, search: String): UserConnection! order(id: ID!): Order } # Mutation root type Mutation { createUser(input: CreateUserInput!): CreateUserPayload! updateUser(input: UpdateUserInput!): UpdateUserPayload! deleteUser(id: ID!): DeleteUserPayload! createOrder(input: CreateOrderInput!): CreateOrderPayload! } # Input types for mutations input CreateUserInput { email: String! name: String! password: String! } # Payload types for mutations type CreateUserPayload { user: User errors: [Error!] } type Error { field: String message: String! } ``` ### Pattern 2: Resolver Design ```python from typing import Optional, List from ariadne import QueryType, MutationType, ObjectType from dataclasses import dataclass query = QueryType() mutation = MutationType() user_type = ObjectType("User") @query.field("user") async def resolve_user(obj, info, id: str) -> Optional[dict]: """Resolve single user by ID.""" return await fetch_user_by_id(id) @query.field("users") async def resolve_users( obj, info, first: int = 20, after: Optional[str] = None, search: Optional[str] = None ) -> dict: """Resolve paginated user list.""" # Decode cursor offset = decode_cursor(after) if after else 0 # Fetch users users = await fetch_users( limit=first + 1, # Fetch one extra to check hasNextPage offset=offset, search=search ) # Pagination has_next = len(users) > first if has_next: users = users[:first] edges = [ { "node": user, "cursor": encode_cursor(offset + i) } for i, user in enumerate(users) ] return { "edges": edges, "pageInfo": { "hasNextPage": has_next, "hasPreviousPage": offset > 0, "startCursor": edges[0]["cursor"] if edges else None, "endCursor": edges[-1]["cursor"] if edges else None }, "totalCount": await count_users(search=search) } @user_type.field("orders") async def resolve_user_orders(user: dict, info, first: int = 20) -> dict: """Resolve user's orders (N+1 prevention with DataLoader).""" # Use DataLoader to batch requests loader = info.context["loaders"]["orders_by_user"] orders = await loader.load(user["id"]) return paginate_orders(orders, first) @mutation.field("createUser") async def resolve_create_user(obj, info, input: dict) -> dict: """Create new user.""" try: # Validate input validate_user_input(input) # Create user user = await create_user( email=input["email"], name=input["name"], password=hash_password(input["password"]) ) return { "user": user, "errors": [] } except ValidationError as e: return { "user": None, "errors": [{"field": e.field, "message": e.message}] } ``` ### Pattern 3: DataLoader (N+1 Problem Prevention) ```python from aiodataloader import DataLoader from typing import List, Optional class UserLoader(DataLoader): """Batch load users by ID.""" async def batch_load_fn(self, user_ids: List[str]) -> List[Optional[dict]]: """Load multiple users in single query.""" users = await fetch_users_by_ids(user_ids) # Map results back to input order user_map = {user["id"]: user for user in users} return [user_map.get(user_id) for user_id in user_ids] class OrdersByUserLoader(DataLoader): """Batch load orders by user ID.""" async def batch_load_fn(self, user_ids: List[str]) -> List[List[dict]]: """Load orders for multiple users in single query.""" orders = await fetch_orders_by_user_ids(user_ids) # Group orders by user_id orders_by_user = {} for order in orders: user_id = order["user_id"] if user_id not in orders_by_user: orders_by_user[user_id] = [] orders_by_user[user_id].append(order) # Return in input order return [orders_by_user.get(user_id, []) for user_id in user_ids] # Context setup def create_context(): return { "loaders": { "user": UserLoader(), "orders_by_user": OrdersByUserLoader() } } ``` ## Best Practices ### REST APIs 1. **Consistent Naming**: Use plural nouns for collections (`/users`, not `/user`) 2. **Stateless**: Each request contains all necessary information 3. **Use HTTP Status Codes Correctly**: 2xx success, 4xx client errors, 5xx server errors 4. **Version Your API**: Plan for breaking changes from day one 5. **Pagination**: Always paginate large collections 6. **Rate Limiting**: Protect your API with rate limits 7. **Documentation**: Use OpenAPI/Swagger for interactive docs ### GraphQL APIs 1. **Schema First**: Design schema before writing resolvers 2. **Avoid N+1**: Use DataLoaders for efficient data fetching 3. **Input Validation**: Validate at schema and resolver levels 4. **Error Handling**: Return structured errors in mutation payloads 5. **Pagination**: Use cursor-based pagination (Relay spec) 6. **Deprecation**: Use `@deprecated` directive for gradual migration 7. **Monitoring**: Track query complexity and execution time ## Common Pitfalls - **Over-fetching/Under-fetching (REST)**: Fixed in GraphQL but requires DataLoaders - **Breaking Changes**: Version APIs or use deprecation strategies - **Inconsistent Error Formats**: Standardize error responses - **Missing Rate Limits**: APIs without limits are vulnerable to abuse - **Poor Documentation**: Undocumented APIs frustrate developers - **Ignoring HTTP Semantics**: POST for idempotent operations breaks expectations - **Tight Coupling**: API structure shouldn't mirror database schema ## Resources - **references/rest-best-practices.md**: Comprehensive REST API design guide - **references/graphql-schema-design.md**: GraphQL schema patterns and anti-patterns - **references/api-versioning-strategies.md**: Versioning approaches and migration paths - **assets/rest-api-template.py**: FastAPI REST API template - **assets/graphql-schema-template.graphql**: Complete GraphQL schema example - **assets/api-design-checklist.md**: Pre-implementation review checklist - **scripts/openapi-generator.py**: Generate OpenAPI specs from code
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

architecture-patterns

Implement proven backend architecture patterns including Clean

coding
⭐1
# Architecture Patterns Master proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, and Domain-Driven Design to build maintainable, testable, and scalable systems. ## When to Use This Skill - Designing new backend systems from scratch - Refactoring monolithic applications for better maintainability - Establishing architecture standards for your team - Migrating from tightly coupled to loosely coupled architectures - Implementing domain-driven design principles - Creating testable and mockable codebases - Planning microservices decomposition ## Core Concepts ### 1. Clean Architecture (Uncle Bob) **Layers (dependency flows inward):** - **Entities**: Core business models - **Use Cases**: Application business rules - **Interface Adapters**: Controllers, presenters, gateways - **Frameworks & Drivers**: UI, database, external services **Key Principles:** - Dependencies point inward - Inner layers know nothing about outer layers - Business logic independent of frameworks - Testable without UI, database, or external services ### 2. Hexagonal Architecture (Ports and Adapters) **Components:** - **Domain Core**: Business logic - **Ports**: Interfaces defining interactions - **Adapters**: Implementations of ports (database, REST, message queue) **Benefits:** - Swap implementations easily (mock for testing) - Technology-agnostic core - Clear separation of concerns ### 3. Domain-Driven Design (DDD) **Strategic Patterns:** - **Bounded Contexts**: Separate models for different domains - **Context Mapping**: How contexts relate - **Ubiquitous Language**: Shared terminology **Tactical Patterns:** - **Entities**: Objects with identity - **Value Objects**: Immutable objects defined by attributes - **Aggregates**: Consistency boundaries - **Repositories**: Data access abstraction - **Domain Events**: Things that happened ## Clean Architecture Pattern ### Directory Structure ``` app/ ├── domain/ # Entities & business rules │ ├── entities/ │ │ ├── user.py │ │ └── order.py │ ├── value_objects/ │ │ ├── email.py │ │ └── money.py │ └── interfaces/ # Abstract interfaces │ ├── user_repository.py │ └── payment_gateway.py ├── use_cases/ # Application business rules │ ├── create_user.py │ ├── process_order.py │ └── send_notification.py ├── adapters/ # Interface implementations │ ├── repositories/ │ │ ├── postgres_user_repository.py │ │ └── redis_cache_repository.py │ ├── controllers/ │ │ └── user_controller.py │ └── gateways/ │ ├── stripe_payment_gateway.py │ └── sendgrid_email_gateway.py └── infrastructure/ # Framework & external concerns ├── database.py ├── config.py └── logging.py ``` ### Implementation Example ```python # domain/entities/user.py from dataclasses import dataclass from datetime import datetime from typing import Optional @dataclass class User: """Core user entity - no framework dependencies.""" id: str email: str name: str created_at: datetime is_active: bool = True def deactivate(self): """Business rule: deactivating user.""" self.is_active = False def can_place_order(self) -> bool: """Business rule: active users can order.""" return self.is_active # domain/interfaces/user_repository.py from abc import ABC, abstractmethod from typing import Optional, List from domain.entities.user import User class IUserRepository(ABC): """Port: defines contract, no implementation.""" @abstractmethod async def find_by_id(self, user_id: str) -> Optional[User]: pass @abstractmethod async def find_by_email(self, email: str) -> Optional[User]: pass @abstractmethod async def save(self, user: User) -> User: pass @abstractmethod async def delete(self, user_id: str) -> bool: pass # use_cases/create_user.py from domain.entities.user import User from domain.interfaces.user_repository import IUserRepository from dataclasses import dataclass from datetime import datetime import uuid @dataclass class CreateUserRequest: email: str name: str @dataclass class CreateUserResponse: user: User success: bool error: Optional[str] = None class CreateUserUseCase: """Use case: orchestrates business logic.""" def __init__(self, user_repository: IUserRepository): self.user_repository = user_repository async def execute(self, request: CreateUserRequest) -> CreateUserResponse: # Business validation existing = await self.user_repository.find_by_email(request.email) if existing: return CreateUserResponse( user=None, success=False, error="Email already exists" ) # Create entity user = User( id=str(uuid.uuid4()), email=request.email, name=request.name, created_at=datetime.now(), is_active=True ) # Persist saved_user = await self.user_repository.save(user) return CreateUserResponse( user=saved_user, success=True ) # adapters/repositories/postgres_user_repository.py from domain.interfaces.user_repository import IUserRepository from domain.entities.user import User from typing import Optional import asyncpg class PostgresUserRepository(IUserRepository): """Adapter: PostgreSQL implementation.""" def __init__(self, pool: asyncpg.Pool): self.pool = pool async def find_by_id(self, user_id: str) -> Optional[User]: async with self.pool.acquire() as conn: row = await conn.fetchrow( "SELECT * FROM users WHERE id = $1", user_id ) return self._to_entity(row) if row else None async def find_by_email(self, email: str) -> Optional[User]: async with self.pool.acquire() as conn: row = await conn.fetchrow( "SELECT * FROM users WHERE email = $1", email ) return self._to_entity(row) if row else None async def save(self, user: User) -> User: async with self.pool.acquire() as conn: await conn.execute( """ INSERT INTO users (id, email, name, created_at, is_active) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (id) DO UPDATE SET email = $2, name = $3, is_active = $5 """, user.id, user.email, user.name, user.created_at, user.is_active ) return user async def delete(self, user_id: str) -> bool: async with self.pool.acquire() as conn: result = await conn.execute( "DELETE FROM users WHERE id = $1", user_id ) return result == "DELETE 1" def _to_entity(self, row) -> User: """Map database row to entity.""" return User( id=row["id"], email=row["email"], name=row["name"], created_at=row["created_at"], is_active=row["is_active"] ) # adapters/controllers/user_controller.py from fastapi import APIRouter, Depends, HTTPException from use_cases.create_user import CreateUserUseCase, CreateUserRequest from pydantic import BaseModel router = APIRouter() class CreateUserDTO(BaseModel): email: str name: str @router.post("/users") async def create_user( dto: CreateUserDTO, use_case: CreateUserUseCase = Depends(get_create_user_use_case) ): """Controller: handles HTTP concerns only.""" request = CreateUserRequest(email=dto.email, name=dto.name) response = await use_case.execute(request) if not response.success: raise HTTPException(status_code=400, detail=response.error) return {"user": response.user} ``` ## Hexagonal Architecture Pattern ```python # Core domain (hexagon center) class OrderService: """Domain service - no infrastructure dependencies.""" def __init__( self, order_repository: OrderRepositoryPort, payment_gateway: PaymentGatewayPort, notification_service: NotificationPort ): self.orders = order_repository self.payments = payment_gateway self.notifications = notification_service async def place_order(self, order: Order) -> OrderResult: # Business logic if not order.is_valid(): return OrderResult(success=False, error="Invalid order") # Use ports (interfaces) payment = await self.payments.charge( amount=order.total, customer=order.customer_id ) if not payment.success: return OrderResult(success=False, error="Payment failed") order.mark_as_paid() saved_order = await self.orders.save(order) await self.notifications.send( to=order.customer_email, subject="Order confirmed", body=f"Order {order.id} confirmed" ) return OrderResult(success=True, order=saved_order) # Ports (interfaces) class OrderRepositoryPort(ABC): @abstractmethod async def save(self, order: Order) -> Order: pass class PaymentGatewayPort(ABC): @abstractmethod async def charge(self, amount: Money, customer: str) -> PaymentResult: pass class NotificationPort(ABC): @abstractmethod async def send(self, to: str, subject: str, body: str): pass # Adapters (implementations) class StripePaymentAdapter(PaymentGatewayPort): """Primary adapter: connects to Stripe API.""" def __init__(self, api_key: str): self.stripe = stripe self.stripe.api_key = api_key async def charge(self, amount: Money, customer: str) -> PaymentResult: try: charge = self.stripe.Charge.create( amount=amount.cents, currency=amount.currency, customer=customer ) return PaymentResult(success=True, transaction_id=charge.id) except stripe.error.CardError as e: return PaymentResult(success=False, error=str(e)) class MockPaymentAdapter(PaymentGatewayPort): """Test adapter: no external dependencies.""" async def charge(self, amount: Money, customer: str) -> PaymentResult: return PaymentResult(success=True, transaction_id="mock-123") ``` ## Domain-Driven Design Pattern ```python # Value Objects (immutable) from dataclasses import dataclass from typing import Optional @dataclass(frozen=True) class Email: """Value object: validated email.""" value: str def __post_init__(self): if "@" not in self.value: raise ValueError("Invalid email") @dataclass(frozen=True) class Money: """Value object: amount with currency.""" amount: int # cents currency: str def add(self, other: "Money") -> "Money": if self.currency != other.currency: raise ValueError("Currency mismatch") return Money(self.amount + other.amount, self.currency) # Entities (with identity) class Order: """Entity: has identity, mutable state.""" def __init__(self, id: str, customer: Customer): self.id = id self.customer = customer self.items: List[OrderItem] = [] self.status = OrderStatus.PENDING self._events: List[DomainEvent] = [] def add_item(self, product: Product, quantity: int): """Business logic in entity.""" item = OrderItem(product, quantity) self.items.append(item) self._events.append(ItemAddedEvent(self.id, item)) def total(self) -> Money: """Calculated property.""" return sum(item.subtotal() for item in self.items) def submit(self): """State transition with business rules.""" if not self.items: raise ValueError("Cannot submit empty order") if self.status != OrderStatus.PENDING: raise ValueError("Order already submitted") self.status = OrderStatus.SUBMITTED self._events.append(OrderSubmittedEvent(self.id)) # Aggregates (consistency boundary) class Customer: """Aggregate root: controls access to entities.""" def __init__(self, id: str, email: Email): self.id = id self.email = email self._addresses: List[Address] = [] self._orders: List[str] = [] # Order IDs, not full objects def add_address(self, address: Address): """Aggregate enforces invariants.""" if len(self._addresses) >= 5: raise ValueError("Maximum 5 addresses allowed") self._addresses.append(address) @property def primary_address(self) -> Optional[Address]: return next((a for a in self._addresses if a.is_primary), None) # Domain Events @dataclass class OrderSubmittedEvent: order_id: str occurred_at: datetime = field(default_factory=datetime.now) # Repository (aggregate persistence) class OrderRepository: """Repository: persist/retrieve aggregates.""" async def find_by_id(self, order_id: str) -> Optional[Order]: """Reconstitute aggregate from storage.""" pass async def save(self, order: Order): """Persist aggregate and publish events.""" await self._persist(order) await self._publish_events(order._events) order._events.clear() ``` ## Resources - **references/clean-architecture-guide.md**: Detailed layer breakdown - **references/hexagonal-architecture-guide.md**: Ports and adapters patterns - **references/ddd-tactical-patterns.md**: Entities, value objects, aggregates - **assets/clean-architecture-template/**: Complete project structure - **assets/ddd-examples/**: Domain modeling examples ## Best Practices 1. **Dependency Rule**: Dependencies always point inward 2. **Interface Segregation**: Small, focused interfaces 3. **Business Logic in Domain**: Keep frameworks out of core 4. **Test Independence**: Core testable without infrastructure 5. **Bounded Contexts**: Clear domain boundaries 6. **Ubiquitous Language**: Consistent terminology 7. **Thin Controllers**: Delegate to use cases 8. **Rich Domain Models**: Behavior with data ## Common Pitfalls - **Anemic Domain**: Entities with only data, no behavior - **Framework Coupling**: Business logic depends on frameworks - **Fat Controllers**: Business logic in controllers - **Repository Leakage**: Exposing ORM objects - **Missing Abstractions**: Concrete dependencies in core - **Over-Engineering**: Clean architecture for simple CRUD
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

cqrs-implementation

Implement Command Query Responsibility Segregation for scalable

coding
⭐1
# CQRS Implementation Comprehensive guide to implementing CQRS (Command Query Responsibility Segregation) patterns. ## When to Use This Skill - Separating read and write concerns - Scaling reads independently from writes - Building event-sourced systems - Optimizing complex query scenarios - Different read/write data models needed - High-performance reporting requirements ## Core Concepts ### 1. CQRS Architecture ``` ┌─────────────┐ │ Client │ └──────┬──────┘ │ ┌────────────┴────────────┐ │ │ ▼ ▼ ┌─────────────┐ ┌─────────────┐ │ Commands │ │ Queries │ │ API │ │ API │ └──────┬──────┘ └──────┬──────┘ │ │ ▼ ▼ ┌─────────────┐ ┌─────────────┐ │ Command │ │ Query │ │ Handlers │ │ Handlers │ └──────┬──────┘ └──────┬──────┘ │ │ ▼ ▼ ┌─────────────┐ ┌─────────────┐ │ Write │─────────►│ Read │ │ Model │ Events │ Model │ └─────────────┘ └─────────────┘ ``` ### 2. Key Components | Component | Responsibility | | ------------------- | ------------------------------- | | **Command** | Intent to change state | | **Command Handler** | Validates and executes commands | | **Event** | Record of state change | | **Query** | Request for data | | **Query Handler** | Retrieves data from read model | | **Projector** | Updates read model from events | ## Templates ### Template 1: Command Infrastructure ```python from abc import ABC, abstractmethod from dataclasses import dataclass from typing import TypeVar, Generic, Dict, Any, Type from datetime import datetime import uuid # Command base @dataclass class Command: command_id: str = None timestamp: datetime = None def __post_init__(self): self.command_id = self.command_id or str(uuid.uuid4()) self.timestamp = self.timestamp or datetime.utcnow() # Concrete commands @dataclass class CreateOrder(Command): customer_id: str items: list shipping_address: dict @dataclass class AddOrderItem(Command): order_id: str product_id: str quantity: int price: float @dataclass class CancelOrder(Command): order_id: str reason: str # Command handler base T = TypeVar('T', bound=Command) class CommandHandler(ABC, Generic[T]): @abstractmethod async def handle(self, command: T) -> Any: pass # Command bus class CommandBus: def __init__(self): self._handlers: Dict[Type[Command], CommandHandler] = {} def register(self, command_type: Type[Command], handler: CommandHandler): self._handlers[command_type] = handler async def dispatch(self, command: Command) -> Any: handler = self._handlers.get(type(command)) if not handler: raise ValueError(f"No handler for {type(command).__name__}") return await handler.handle(command) # Command handler implementation class CreateOrderHandler(CommandHandler[CreateOrder]): def __init__(self, order_repository, event_store): self.order_repository = order_repository self.event_store = event_store async def handle(self, command: CreateOrder) -> str: # Validate if not command.items: raise ValueError("Order must have at least one item") # Create aggregate order = Order.create( customer_id=command.customer_id, items=command.items, shipping_address=command.shipping_address ) # Persist events await self.event_store.append_events( stream_id=f"Order-{order.id}", stream_type="Order", events=order.uncommitted_events ) return order.id ``` ### Template 2: Query Infrastructure ```python from abc import ABC, abstractmethod from dataclasses import dataclass from typing import TypeVar, Generic, List, Optional # Query base @dataclass class Query: pass # Concrete queries @dataclass class GetOrderById(Query): order_id: str @dataclass class GetCustomerOrders(Query): customer_id: str status: Optional[str] = None page: int = 1 page_size: int = 20 @dataclass class SearchOrders(Query): query: str filters: dict = None sort_by: str = "created_at" sort_order: str = "desc" # Query result types @dataclass class OrderView: order_id: str customer_id: str status: str total_amount: float item_count: int created_at: datetime shipped_at: Optional[datetime] = None @dataclass class PaginatedResult(Generic[T]): items: List[T] total: int page: int page_size: int @property def total_pages(self) -> int: return (self.total + self.page_size - 1) // self.page_size # Query handler base T = TypeVar('T', bound=Query) R = TypeVar('R') class QueryHandler(ABC, Generic[T, R]): @abstractmethod async def handle(self, query: T) -> R: pass # Query bus class QueryBus: def __init__(self): self._handlers: Dict[Type[Query], QueryHandler] = {} def register(self, query_type: Type[Query], handler: QueryHandler): self._handlers[query_type] = handler async def dispatch(self, query: Query) -> Any: handler = self._handlers.get(type(query)) if not handler: raise ValueError(f"No handler for {type(query).__name__}") return await handler.handle(query) # Query handler implementation class GetOrderByIdHandler(QueryHandler[GetOrderById, Optional[OrderView]]): def __init__(self, read_db): self.read_db = read_db async def handle(self, query: GetOrderById) -> Optional[OrderView]: async with self.read_db.acquire() as conn: row = await conn.fetchrow( """ SELECT order_id, customer_id, status, total_amount, item_count, created_at, shipped_at FROM order_views WHERE order_id = $1 """, query.order_id ) if row: return OrderView(**dict(row)) return None class GetCustomerOrdersHandler(QueryHandler[GetCustomerOrders, PaginatedResult[OrderView]]): def __init__(self, read_db): self.read_db = read_db async def handle(self, query: GetCustomerOrders) -> PaginatedResult[OrderView]: async with self.read_db.acquire() as conn: # Build query with optional status filter where_clause = "customer_id = $1" params = [query.customer_id] if query.status: where_clause += " AND status = $2" params.append(query.status) # Get total count total = await conn.fetchval( f"SELECT COUNT(*) FROM order_views WHERE {where_clause}", *params ) # Get paginated results offset = (query.page - 1) * query.page_size rows = await conn.fetch( f""" SELECT order_id, customer_id, status, total_amount, item_count, created_at, shipped_at FROM order_views WHERE {where_clause} ORDER BY created_at DESC LIMIT ${len(params) + 1} OFFSET ${len(params) + 2} """, *params, query.page_size, offset ) return PaginatedResult( items=[OrderView(**dict(row)) for row in rows], total=total, page=query.page, page_size=query.page_size ) ``` ### Template 3: FastAPI CQRS Application ```python from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from typing import List, Optional app = FastAPI() # Request/Response models class CreateOrderRequest(BaseModel): customer_id: str items: List[dict] shipping_address: dict class OrderResponse(BaseModel): order_id: str customer_id: str status: str total_amount: float item_count: int created_at: datetime # Dependency injection def get_command_bus() -> CommandBus: return app.state.command_bus def get_query_bus() -> QueryBus: return app.state.query_bus # Command endpoints (POST, PUT, DELETE) @app.post("/orders", response_model=dict) async def create_order( request: CreateOrderRequest, command_bus: CommandBus = Depends(get_command_bus) ): command = CreateOrder( customer_id=request.customer_id, items=request.items, shipping_address=request.shipping_address ) order_id = await command_bus.dispatch(command) return {"order_id": order_id} @app.post("/orders/{order_id}/items") async def add_item( order_id: str, product_id: str, quantity: int, price: float, command_bus: CommandBus = Depends(get_command_bus) ): command = AddOrderItem( order_id=order_id, product_id=product_id, quantity=quantity, price=price ) await command_bus.dispatch(command) return {"status": "item_added"} @app.delete("/orders/{order_id}") async def cancel_order( order_id: str, reason: str, command_bus: CommandBus = Depends(get_command_bus) ): command = CancelOrder(order_id=order_id, reason=reason) await command_bus.dispatch(command) return {"status": "cancelled"} # Query endpoints (GET) @app.get("/orders/{order_id}", response_model=OrderResponse) async def get_order( order_id: str, query_bus: QueryBus = Depends(get_query_bus) ): query = GetOrderById(order_id=order_id) result = await query_bus.dispatch(query) if not result: raise HTTPException(status_code=404, detail="Order not found") return result @app.get("/customers/{customer_id}/orders") async def get_customer_orders( customer_id: str, status: Optional[str] = None, page: int = 1, page_size: int = 20, query_bus: QueryBus = Depends(get_query_bus) ): query = GetCustomerOrders( customer_id=customer_id, status=status, page=page, page_size=page_size ) return await query_bus.dispatch(query) @app.get("/orders/search") async def search_orders( q: str, sort_by: str = "created_at", query_bus: QueryBus = Depends(get_query_bus) ): query = SearchOrders(query=q, sort_by=sort_by) return await query_bus.dispatch(query) ``` ### Template 4: Read Model Synchronization ```python class ReadModelSynchronizer: """Keeps read models in sync with events.""" def __init__(self, event_store, read_db, projections: List[Projection]): self.event_store = event_store self.read_db = read_db self.projections = {p.name: p for p in projections} async def run(self): """Continuously sync read models.""" while True: for name, projection in self.projections.items(): await self._sync_projection(projection) await asyncio.sleep(0.1) async def _sync_projection(self, projection: Projection): checkpoint = await self._get_checkpoint(projection.name) events = await self.event_store.read_all( from_position=checkpoint, limit=100 ) for event in events: if event.event_type in projection.handles(): try: await projection.apply(event) except Exception as e: # Log error, possibly retry or skip logger.error(f"Projection error: {e}") continue await self._save_checkpoint(projection.name, event.global_position) async def rebuild_projection(self, projection_name: str): """Rebuild a projection from scratch.""" projection = self.projections[projection_name] # Clear existing data await projection.clear() # Reset checkpoint await self._save_checkpoint(projection_name, 0) # Rebuild while True: checkpoint = await self._get_checkpoint(projection_name) events = await self.event_store.read_all(checkpoint, 1000) if not events: break for event in events: if event.event_type in projection.handles(): await projection.apply(event) await self._save_checkpoint( projection_name, events[-1].global_position ) ``` ### Template 5: Eventual Consistency Handling ```python class ConsistentQueryHandler: """Query handler that can wait for consistency.""" def __init__(self, read_db, event_store): self.read_db = read_db self.event_store = event_store async def query_after_command( self, query: Query, expected_version: int, stream_id: str, timeout: float = 5.0 ): """ Execute query, ensuring read model is at expected version. Used for read-your-writes consistency. """ start_time = time.time() while time.time() - start_time < timeout: # Check if read model is caught up projection_version = await self._get_projection_version(stream_id) if projection_version >= expected_version: return await self.execute_query(query) # Wait a bit and retry await asyncio.sleep(0.1) # Timeout - return stale data with warning return { "data": await self.execute_query(query), "_warning": "Data may be stale" } async def _get_projection_version(self, stream_id: str) -> int: """Get the last processed event version for a stream.""" async with self.read_db.acquire() as conn: return await conn.fetchval( "SELECT last_event_version FROM projection_state WHERE stream_id = $1", stream_id ) or 0 ``` ## Best Practices ### Do's - **Separate command and query models** - Different needs - **Use eventual consistency** - Accept propagation delay - **Validate in command handlers** - Before state change - **Denormalize read models** - Optimize for queries - **Version your events** - For schema evolution ### Don'ts - **Don't query in commands** - Use only for writes - **Don't couple read/write schemas** - Independent evolution - **Don't over-engineer** - Start simple - **Don't ignore consistency SLAs** - Define acceptable lag ## Resources - [CQRS Pattern](https://martinfowler.com/bliki/CQRS.html) - [Microsoft CQRS Guidance](https://docs.microsoft.com/en-us/azure/architecture/patterns/cqrs)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

microservices-patterns

Design microservices architectures with service boundaries,

coding
⭐1
# Microservices Patterns Master microservices architecture patterns including service boundaries, inter-service communication, data management, and resilience patterns for building distributed systems. ## When to Use This Skill - Decomposing monoliths into microservices - Designing service boundaries and contracts - Implementing inter-service communication - Managing distributed data and transactions - Building resilient distributed systems - Implementing service discovery and load balancing - Designing event-driven architectures ## Core Concepts ### 1. Service Decomposition Strategies **By Business Capability** - Organize services around business functions - Each service owns its domain - Example: OrderService, PaymentService, InventoryService **By Subdomain (DDD)** - Core domain, supporting subdomains - Bounded contexts map to services - Clear ownership and responsibility **Strangler Fig Pattern** - Gradually extract from monolith - New functionality as microservices - Proxy routes to old/new systems ### 2. Communication Patterns **Synchronous (Request/Response)** - REST APIs - gRPC - GraphQL **Asynchronous (Events/Messages)** - Event streaming (Kafka) - Message queues (RabbitMQ, SQS) - Pub/Sub patterns ### 3. Data Management **Database Per Service** - Each service owns its data - No shared databases - Loose coupling **Saga Pattern** - Distributed transactions - Compensating actions - Eventual consistency ### 4. Resilience Patterns **Circuit Breaker** - Fail fast on repeated errors - Prevent cascade failures **Retry with Backoff** - Transient fault handling - Exponential backoff **Bulkhead** - Isolate resources - Limit impact of failures ## Service Decomposition Patterns ### Pattern 1: By Business Capability ```python # E-commerce example # Order Service class OrderService: """Handles order lifecycle.""" async def create_order(self, order_data: dict) -> Order: order = Order.create(order_data) # Publish event for other services await self.event_bus.publish( OrderCreatedEvent( order_id=order.id, customer_id=order.customer_id, items=order.items, total=order.total ) ) return order # Payment Service (separate service) class PaymentService: """Handles payment processing.""" async def process_payment(self, payment_request: PaymentRequest) -> PaymentResult: # Process payment result = await self.payment_gateway.charge( amount=payment_request.amount, customer=payment_request.customer_id ) if result.success: await self.event_bus.publish( PaymentCompletedEvent( order_id=payment_request.order_id, transaction_id=result.transaction_id ) ) return result # Inventory Service (separate service) class InventoryService: """Handles inventory management.""" async def reserve_items(self, order_id: str, items: List[OrderItem]) -> ReservationResult: # Check availability for item in items: available = await self.inventory_repo.get_available(item.product_id) if available < item.quantity: return ReservationResult( success=False, error=f"Insufficient inventory for {item.product_id}" ) # Reserve items reservation = await self.create_reservation(order_id, items) await self.event_bus.publish( InventoryReservedEvent( order_id=order_id, reservation_id=reservation.id ) ) return ReservationResult(success=True, reservation=reservation) ``` ### Pattern 2: API Gateway ```python from fastapi import FastAPI, HTTPException, Depends import httpx from circuitbreaker import circuit app = FastAPI() class APIGateway: """Central entry point for all client requests.""" def __init__(self): self.order_service_url = "http://order-service:8000" self.payment_service_url = "http://payment-service:8001" self.inventory_service_url = "http://inventory-service:8002" self.http_client = httpx.AsyncClient(timeout=5.0) @circuit(failure_threshold=5, recovery_timeout=30) async def call_order_service(self, path: str, method: str = "GET", **kwargs): """Call order service with circuit breaker.""" response = await self.http_client.request( method, f"{self.order_service_url}{path}", **kwargs ) response.raise_for_status() return response.json() async def create_order_aggregate(self, order_id: str) -> dict: """Aggregate data from multiple services.""" # Parallel requests order, payment, inventory = await asyncio.gather( self.call_order_service(f"/orders/{order_id}"), self.call_payment_service(f"/payments/order/{order_id}"), self.call_inventory_service(f"/reservations/order/{order_id}"), return_exceptions=True ) # Handle partial failures result = {"order": order} if not isinstance(payment, Exception): result["payment"] = payment if not isinstance(inventory, Exception): result["inventory"] = inventory return result @app.post("/api/orders") async def create_order( order_data: dict, gateway: APIGateway = Depends() ): """API Gateway endpoint.""" try: # Route to order service order = await gateway.call_order_service( "/orders", method="POST", json=order_data ) return {"order": order} except httpx.HTTPError as e: raise HTTPException(status_code=503, detail="Order service unavailable") ``` ## Communication Patterns ### Pattern 1: Synchronous REST Communication ```python # Service A calls Service B import httpx from tenacity import retry, stop_after_attempt, wait_exponential class ServiceClient: """HTTP client with retries and timeout.""" def __init__(self, base_url: str): self.base_url = base_url self.client = httpx.AsyncClient( timeout=httpx.Timeout(5.0, connect=2.0), limits=httpx.Limits(max_keepalive_connections=20) ) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10) ) async def get(self, path: str, **kwargs): """GET with automatic retries.""" response = await self.client.get(f"{self.base_url}{path}", **kwargs) response.raise_for_status() return response.json() async def post(self, path: str, **kwargs): """POST request.""" response = await self.client.post(f"{self.base_url}{path}", **kwargs) response.raise_for_status() return response.json() # Usage payment_client = ServiceClient("http://payment-service:8001") result = await payment_client.post("/payments", json=payment_data) ``` ### Pattern 2: Asynchronous Event-Driven ```python # Event-driven communication with Kafka from aiokafka import AIOKafkaProducer, AIOKafkaConsumer import json from dataclasses import dataclass, asdict from datetime import datetime @dataclass class DomainEvent: event_id: str event_type: str aggregate_id: str occurred_at: datetime data: dict class EventBus: """Event publishing and subscription.""" def __init__(self, bootstrap_servers: List[str]): self.bootstrap_servers = bootstrap_servers self.producer = None async def start(self): self.producer = AIOKafkaProducer( bootstrap_servers=self.bootstrap_servers, value_serializer=lambda v: json.dumps(v).encode() ) await self.producer.start() async def publish(self, event: DomainEvent): """Publish event to Kafka topic.""" topic = event.event_type await self.producer.send_and_wait( topic, value=asdict(event), key=event.aggregate_id.encode() ) async def subscribe(self, topic: str, handler: callable): """Subscribe to events.""" consumer = AIOKafkaConsumer( topic, bootstrap_servers=self.bootstrap_servers, value_deserializer=lambda v: json.loads(v.decode()), group_id="my-service" ) await consumer.start() try: async for message in consumer: event_data = message.value await handler(event_data) finally: await consumer.stop() # Order Service publishes event async def create_order(order_data: dict): order = await save_order(order_data) event = DomainEvent( event_id=str(uuid.uuid4()), event_type="OrderCreated", aggregate_id=order.id, occurred_at=datetime.now(), data={ "order_id": order.id, "customer_id": order.customer_id, "total": order.total } ) await event_bus.publish(event) # Inventory Service listens for OrderCreated async def handle_order_created(event_data: dict): """React to order creation.""" order_id = event_data["data"]["order_id"] items = event_data["data"]["items"] # Reserve inventory await reserve_inventory(order_id, items) ``` ### Pattern 3: Saga Pattern (Distributed Transactions) ```python # Saga orchestration for order fulfillment from enum import Enum from typing import List, Callable class SagaStep: """Single step in saga.""" def __init__( self, name: str, action: Callable, compensation: Callable ): self.name = name self.action = action self.compensation = compensation class SagaStatus(Enum): PENDING = "pending" COMPLETED = "completed" COMPENSATING = "compensating" FAILED = "failed" class OrderFulfillmentSaga: """Orchestrated saga for order fulfillment.""" def __init__(self): self.steps: List[SagaStep] = [ SagaStep( "create_order", action=self.create_order, compensation=self.cancel_order ), SagaStep( "reserve_inventory", action=self.reserve_inventory, compensation=self.release_inventory ), SagaStep( "process_payment", action=self.process_payment, compensation=self.refund_payment ), SagaStep( "confirm_order", action=self.confirm_order, compensation=self.cancel_order_confirmation ) ] async def execute(self, order_data: dict) -> SagaResult: """Execute saga steps.""" completed_steps = [] context = {"order_data": order_data} try: for step in self.steps: # Execute step result = await step.action(context) if not result.success: # Compensate await self.compensate(completed_steps, context) return SagaResult( status=SagaStatus.FAILED, error=result.error ) completed_steps.append(step) context.update(result.data) return SagaResult(status=SagaStatus.COMPLETED, data=context) except Exception as e: # Compensate on error await self.compensate(completed_steps, context) return SagaResult(status=SagaStatus.FAILED, error=str(e)) async def compensate(self, completed_steps: List[SagaStep], context: dict): """Execute compensating actions in reverse order.""" for step in reversed(completed_steps): try: await step.compensation(context) except Exception as e: # Log compensation failure print(f"Compensation failed for {step.name}: {e}") # Step implementations async def create_order(self, context: dict) -> StepResult: order = await order_service.create(context["order_data"]) return StepResult(success=True, data={"order_id": order.id}) async def cancel_order(self, context: dict): await order_service.cancel(context["order_id"]) async def reserve_inventory(self, context: dict) -> StepResult: result = await inventory_service.reserve( context["order_id"], context["order_data"]["items"] ) return StepResult( success=result.success, data={"reservation_id": result.reservation_id} ) async def release_inventory(self, context: dict): await inventory_service.release(context["reservation_id"]) async def process_payment(self, context: dict) -> StepResult: result = await payment_service.charge( context["order_id"], context["order_data"]["total"] ) return StepResult( success=result.success, data={"transaction_id": result.transaction_id}, error=result.error ) async def refund_payment(self, context: dict): await payment_service.refund(context["transaction_id"]) ``` ## Resilience Patterns ### Circuit Breaker Pattern ```python from enum import Enum from datetime import datetime, timedelta from typing import Callable, Any class CircuitState(Enum): CLOSED = "closed" # Normal operation OPEN = "open" # Failing, reject requests HALF_OPEN = "half_open" # Testing if recovered class CircuitBreaker: """Circuit breaker for service calls.""" def __init__( self, failure_threshold: int = 5, recovery_timeout: int = 30, success_threshold: int = 2 ): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.success_threshold = success_threshold self.failure_count = 0 self.success_count = 0 self.state = CircuitState.CLOSED self.opened_at = None async def call(self, func: Callable, *args, **kwargs) -> Any: """Execute function with circuit breaker.""" if self.state == CircuitState.OPEN: if self._should_attempt_reset(): self.state = CircuitState.HALF_OPEN else: raise CircuitBreakerOpenError("Circuit breaker is open") try: result = await func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise def _on_success(self): """Handle successful call.""" self.failure_count = 0 if self.state == CircuitState.HALF_OPEN: self.success_count += 1 if self.success_count >= self.success_threshold: self.state = CircuitState.CLOSED self.success_count = 0 def _on_failure(self): """Handle failed call.""" self.failure_count += 1 if self.failure_count >= self.failure_threshold: self.state = CircuitState.OPEN self.opened_at = datetime.now() if self.state == CircuitState.HALF_OPEN: self.state = CircuitState.OPEN self.opened_at = datetime.now() def _should_attempt_reset(self) -> bool: """Check if enough time passed to try again.""" return ( datetime.now() - self.opened_at > timedelta(seconds=self.recovery_timeout) ) # Usage breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30) async def call_payment_service(payment_data: dict): return await breaker.call( payment_client.process_payment, payment_data ) ``` ## Resources - **references/service-decomposition-guide.md**: Breaking down monoliths - **references/communication-patterns.md**: Sync vs async patterns - **references/saga-implementation.md**: Distributed transactions - **assets/circuit-breaker.py**: Production circuit breaker - **assets/event-bus-template.py**: Kafka event bus implementation - **assets/api-gateway-template.py**: Complete API gateway ## Best Practices 1. **Service Boundaries**: Align with business capabilities 2. **Database Per Service**: No shared databases 3. **API Contracts**: Versioned, backward compatible 4. **Async When Possible**: Events over direct calls 5. **Circuit Breakers**: Fail fast on service failures 6. **Distributed Tracing**: Track requests across services 7. **Service Registry**: Dynamic service discovery 8. **Health Checks**: Liveness and readiness probes ## Common Pitfalls - **Distributed Monolith**: Tightly coupled services - **Chatty Services**: Too many inter-service calls - **Shared Databases**: Tight coupling through data - **No Circuit Breakers**: Cascade failures - **Synchronous Everything**: Tight coupling, poor resilience - **Premature Microservices**: Starting with microservices - **Ignoring Network Failures**: Assuming reliable network - **No Compensation Logic**: Can't undo failed transactions
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

openapi-spec-generation

Generate and maintain OpenAPI 3.1 specifications from code,

coding
⭐1
# OpenAPI Spec Generation Comprehensive patterns for creating, maintaining, and validating OpenAPI 3.1 specifications for RESTful APIs. ## When to Use This Skill - Creating API documentation from scratch - Generating OpenAPI specs from existing code - Designing API contracts (design-first approach) - Validating API implementations against specs - Generating client SDKs from specs - Setting up API documentation portals ## Core Concepts ### 1. OpenAPI 3.1 Structure ```yaml openapi: 3.1.0 info: title: API Title version: 1.0.0 servers: - url: https://api.example.com/v1 paths: /resources: get: ... components: schemas: ... securitySchemes: ... ``` ### 2. Design Approaches | Approach | Description | Best For | | ---------------- | ---------------------------- | ------------------- | | **Design-First** | Write spec before code | New APIs, contracts | | **Code-First** | Generate spec from code | Existing APIs | | **Hybrid** | Annotate code, generate spec | Evolving APIs | ## Templates ### Template 1: Complete API Specification ```yaml openapi: 3.1.0 info: title: User Management API description: | API for managing users and their profiles. ## Authentication All endpoints require Bearer token authentication. ## Rate Limiting - 1000 requests per minute for standard tier - 10000 requests per minute for enterprise tier version: 2.0.0 contact: name: API Support email: api-support@example.com url: https://docs.example.com license: name: MIT url: https://opensource.org/licenses/MIT servers: - url: https://api.example.com/v2 description: Production - url: https://staging-api.example.com/v2 description: Staging - url: http://localhost:3000/v2 description: Local development tags: - name: Users description: User management operations - name: Profiles description: User profile operations - name: Admin description: Administrative operations paths: /users: get: operationId: listUsers summary: List all users description: Returns a paginated list of users with optional filtering. tags: - Users parameters: - $ref: "#/components/parameters/PageParam" - $ref: "#/components/parameters/LimitParam" - name: status in: query description: Filter by user status schema: $ref: "#/components/schemas/UserStatus" - name: search in: query description: Search by name or email schema: type: string minLength: 2 maxLength: 100 responses: "200": description: Successful response content: application/json: schema: $ref: "#/components/schemas/UserListResponse" examples: default: $ref: "#/components/examples/UserListExample" "400": $ref: "#/components/responses/BadRequest" "401": $ref: "#/components/responses/Unauthorized" "429": $ref: "#/components/responses/RateLimited" security: - bearerAuth: [] post: operationId: createUser summary: Create a new user description: Creates a new user account and sends welcome email. tags: - Users requestBody: required: true content: application/json: schema: $ref: "#/components/schemas/CreateUserRequest" examples: standard: summary: Standard user value: email: user@example.com name: John Doe role: user admin: summary: Admin user value: email: admin@example.com name: Admin User role: admin responses: "201": description: User created successfully content: application/json: schema: $ref: "#/components/schemas/User" headers: Location: description: URL of created user schema: type: string format: uri "400": $ref: "#/components/responses/BadRequest" "409": description: Email already exists content: application/json: schema: $ref: "#/components/schemas/Error" security: - bearerAuth: [] /users/{userId}: parameters: - $ref: "#/components/parameters/UserIdParam" get: operationId: getUser summary: Get user by ID tags: - Users responses: "200": description: Successful response content: application/json: schema: $ref: "#/components/schemas/User" "404": $ref: "#/components/responses/NotFound" security: - bearerAuth: [] patch: operationId: updateUser summary: Update user tags: - Users requestBody: required: true content: application/json: schema: $ref: "#/components/schemas/UpdateUserRequest" responses: "200": description: User updated content: application/json: schema: $ref: "#/components/schemas/User" "400": $ref: "#/components/responses/BadRequest" "404": $ref: "#/components/responses/NotFound" security: - bearerAuth: [] delete: operationId: deleteUser summary: Delete user tags: - Users - Admin responses: "204": description: User deleted "404": $ref: "#/components/responses/NotFound" security: - bearerAuth: [] - apiKey: [] components: schemas: User: type: object required: - id - email - name - status - createdAt properties: id: type: string format: uuid readOnly: true description: Unique user identifier email: type: string format: email description: User email address name: type: string minLength: 1 maxLength: 100 description: User display name status: $ref: "#/components/schemas/UserStatus" role: type: string enum: [user, moderator, admin] default: user avatar: type: string format: uri nullable: true metadata: type: object additionalProperties: true description: Custom metadata createdAt: type: string format: date-time readOnly: true updatedAt: type: string format: date-time readOnly: true UserStatus: type: string enum: [active, inactive, suspended, pending] description: User account status CreateUserRequest: type: object required: - email - name properties: email: type: string format: email name: type: string minLength: 1 maxLength: 100 role: type: string enum: [user, moderator, admin] default: user metadata: type: object additionalProperties: true UpdateUserRequest: type: object minProperties: 1 properties: name: type: string minLength: 1 maxLength: 100 status: $ref: "#/components/schemas/UserStatus" role: type: string enum: [user, moderator, admin] metadata: type: object additionalProperties: true UserListResponse: type: object required: - data - pagination properties: data: type: array items: $ref: "#/components/schemas/User" pagination: $ref: "#/components/schemas/Pagination" Pagination: type: object required: - page - limit - total - totalPages properties: page: type: integer minimum: 1 limit: type: integer minimum: 1 maximum: 100 total: type: integer minimum: 0 totalPages: type: integer minimum: 0 hasNext: type: boolean hasPrev: type: boolean Error: type: object required: - code - message properties: code: type: string description: Error code for programmatic handling message: type: string description: Human-readable error message details: type: array items: type: object properties: field: type: string message: type: string requestId: type: string description: Request ID for support parameters: UserIdParam: name: userId in: path required: true description: User ID schema: type: string format: uuid PageParam: name: page in: query description: Page number (1-based) schema: type: integer minimum: 1 default: 1 LimitParam: name: limit in: query description: Items per page schema: type: integer minimum: 1 maximum: 100 default: 20 responses: BadRequest: description: Invalid request content: application/json: schema: $ref: "#/components/schemas/Error" example: code: VALIDATION_ERROR message: Invalid request parameters details: - field: email message: Must be a valid email address Unauthorized: description: Authentication required content: application/json: schema: $ref: "#/components/schemas/Error" example: code: UNAUTHORIZED message: Authentication required NotFound: description: Resource not found content: application/json: schema: $ref: "#/components/schemas/Error" example: code: NOT_FOUND message: User not found RateLimited: description: Too many requests content: application/json: schema: $ref: "#/components/schemas/Error" headers: Retry-After: description: Seconds until rate limit resets schema: type: integer X-RateLimit-Limit: description: Request limit per window schema: type: integer X-RateLimit-Remaining: description: Remaining requests in window schema: type: integer examples: UserListExample: value: data: - id: "550e8400-e29b-41d4-a716-446655440000" email: "john@example.com" name: "John Doe" status: "active" role: "user" createdAt: "2024-01-15T10:30:00Z" pagination: page: 1 limit: 20 total: 1 totalPages: 1 hasNext: false hasPrev: false securitySchemes: bearerAuth: type: http scheme: bearer bearerFormat: JWT description: JWT token from /auth/login apiKey: type: apiKey in: header name: X-API-Key description: API key for service-to-service calls security: - bearerAuth: [] ``` ### Template 2: Code-First Generation (Python/FastAPI) ```python # FastAPI with automatic OpenAPI generation from fastapi import FastAPI, HTTPException, Query, Path, Depends from pydantic import BaseModel, Field, EmailStr from typing import Optional, List from datetime import datetime from uuid import UUID from enum import Enum app = FastAPI( title="User Management API", description="API for managing users and profiles", version="2.0.0", openapi_tags=[ {"name": "Users", "description": "User operations"}, {"name": "Profiles", "description": "Profile operations"}, ], servers=[ {"url": "https://api.example.com/v2", "description": "Production"}, {"url": "http://localhost:8000", "description": "Development"}, ], ) # Enums class UserStatus(str, Enum): active = "active" inactive = "inactive" suspended = "suspended" pending = "pending" class UserRole(str, Enum): user = "user" moderator = "moderator" admin = "admin" # Models class UserBase(BaseModel): email: EmailStr = Field(..., description="User email address") name: str = Field(..., min_length=1, max_length=100, description="Display name") class UserCreate(UserBase): role: UserRole = Field(default=UserRole.user) metadata: Optional[dict] = Field(default=None, description="Custom metadata") model_config = { "json_schema_extra": { "examples": [ { "email": "user@example.com", "name": "John Doe", "role": "user" } ] } } class UserUpdate(BaseModel): name: Optional[str] = Field(None, min_length=1, max_length=100) status: Optional[UserStatus] = None role: Optional[UserRole] = None metadata: Optional[dict] = None class User(UserBase): id: UUID = Field(..., description="Unique identifier") status: UserStatus role: UserRole avatar: Optional[str] = Field(None, description="Avatar URL") metadata: Optional[dict] = None created_at: datetime = Field(..., alias="createdAt") updated_at: Optional[datetime] = Field(None, alias="updatedAt") model_config = {"populate_by_name": True} class Pagination(BaseModel): page: int = Field(..., ge=1) limit: int = Field(..., ge=1, le=100) total: int = Field(..., ge=0) total_pages: int = Field(..., ge=0, alias="totalPages") has_next: bool = Field(..., alias="hasNext") has_prev: bool = Field(..., alias="hasPrev") class UserListResponse(BaseModel): data: List[User] pagination: Pagination class ErrorDetail(BaseModel): field: str message: str class ErrorResponse(BaseModel): code: str = Field(..., description="Error code") message: str = Field(..., description="Error message") details: Optional[List[ErrorDetail]] = None request_id: Optional[str] = Field(None, alias="requestId") # Endpoints @app.get( "/users", response_model=UserListResponse, tags=["Users"], summary="List all users", description="Returns a paginated list of users with optional filtering.", responses={ 400: {"model": ErrorResponse, "description": "Invalid request"}, 401: {"model": ErrorResponse, "description": "Unauthorized"}, }, ) async def list_users( page: int = Query(1, ge=1, description="Page number"), limit: int = Query(20, ge=1, le=100, description="Items per page"), status: Optional[UserStatus] = Query(None, description="Filter by status"), search: Optional[str] = Query(None, min_length=2, max_length=100), ): """ List users with pagination and filtering. - **page**: Page number (1-based) - **limit**: Number of items per page (max 100) - **status**: Filter by user status - **search**: Search by name or email """ # Implementation pass @app.post( "/users", response_model=User, status_code=201, tags=["Users"], summary="Create a new user", responses={ 400: {"model": ErrorResponse}, 409: {"model": ErrorResponse, "description": "Email already exists"}, }, ) async def create_user(user: UserCreate): """Create a new user and send welcome email.""" pass @app.get( "/users/{user_id}", response_model=User, tags=["Users"], summary="Get user by ID", responses={404: {"model": ErrorResponse}}, ) async def get_user( user_id: UUID = Path(..., description="User ID"), ): """Retrieve a specific user by their ID.""" pass @app.patch( "/users/{user_id}", response_model=User, tags=["Users"], summary="Update user", responses={ 400: {"model": ErrorResponse}, 404: {"model": ErrorResponse}, }, ) async def update_user( user_id: UUID = Path(..., description="User ID"), user: UserUpdate = ..., ): """Update user attributes.""" pass @app.delete( "/users/{user_id}", status_code=204, tags=["Users", "Admin"], summary="Delete user", responses={404: {"model": ErrorResponse}}, ) async def delete_user( user_id: UUID = Path(..., description="User ID"), ): """Permanently delete a user.""" pass # Export OpenAPI spec if __name__ == "__main__": import json print(json.dumps(app.openapi(), indent=2)) ``` ### Template 3: Code-First (TypeScript/Express with tsoa) ```typescript // tsoa generates OpenAPI from TypeScript decorators import { Controller, Get, Post, Patch, Delete, Route, Path, Query, Body, Response, SuccessResponse, Tags, Security, Example, } from "tsoa"; // Models interface User { /** Unique identifier */ id: string; /** User email address */ email: string; /** Display name */ name: string; status: UserStatus; role: UserRole; /** Avatar URL */ avatar?: string; /** Custom metadata */ metadata?: Record<string, unknown>; createdAt: Date; updatedAt?: Date; } enum UserStatus { Active = "active", Inactive = "inactive", Suspended = "suspended", Pending = "pending", } enum UserRole { User = "user", Moderator = "moderator", Admin = "admin", } interface CreateUserRequest { email: string; name: string; role?: UserRole; metadata?: Record<string, unknown>; } interface UpdateUserRequest { name?: string; status?: UserStatus; role?: UserRole; metadata?: Record<string, unknown>; } interface Pagination { page: number; limit: number; total: number; totalPages: number; hasNext: boolean; hasPrev: boolean; } interface UserListResponse { data: User[]; pagination: Pagination; } interface ErrorResponse { code: string; message: string; details?: { field: string; message: string }[]; requestId?: string; } @Route("users") @Tags("Users") export class UsersController extends Controller { /** * List all users with pagination and filtering * @param page Page number (1-based) * @param limit Items per page (max 100) * @param status Filter by user status * @param search Search by name or email */
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

async-python-patterns

Master Python asyncio, concurrent programming, and async/await

coding
⭐1
# Async Python Patterns Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems. ## When to Use This Skill - Building async web APIs (FastAPI, aiohttp, Sanic) - Implementing concurrent I/O operations (database, file, network) - Creating web scrapers with concurrent requests - Developing real-time applications (WebSocket servers, chat systems) - Processing multiple independent tasks simultaneously - Building microservices with async communication - Optimizing I/O-bound workloads - Implementing async background tasks and queues ## Sync vs Async Decision Guide Before adopting async, consider whether it's the right choice for your use case. | Use Case | Recommended Approach | |----------|---------------------| | Many concurrent network/DB calls | `asyncio` | | CPU-bound computation | `multiprocessing` or thread pool | | Mixed I/O + CPU | Offload CPU work with `asyncio.to_thread()` | | Simple scripts, few connections | Sync (simpler, easier to debug) | | Web APIs with high concurrency | Async frameworks (FastAPI, aiohttp) | **Key Rule:** Stay fully sync or fully async within a call path. Mixing creates hidden blocking and complexity. ## Core Concepts ### 1. Event Loop The event loop is the heart of asyncio, managing and scheduling asynchronous tasks. **Key characteristics:** - Single-threaded cooperative multitasking - Schedules coroutines for execution - Handles I/O operations without blocking - Manages callbacks and futures ### 2. Coroutines Functions defined with `async def` that can be paused and resumed. **Syntax:** ```python async def my_coroutine(): result = await some_async_operation() return result ``` ### 3. Tasks Scheduled coroutines that run concurrently on the event loop. ### 4. Futures Low-level objects representing eventual results of async operations. ### 5. Async Context Managers Resources that support `async with` for proper cleanup. ### 6. Async Iterators Objects that support `async for` for iterating over async data sources. ## Quick Start ```python import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") # Python 3.7+ asyncio.run(main()) ``` ## Fundamental Patterns ### Pattern 1: Basic Async/Await ```python import asyncio async def fetch_data(url: str) -> dict: """Fetch data from URL asynchronously.""" await asyncio.sleep(1) # Simulate I/O return {"url": url, "data": "result"} async def main(): result = await fetch_data("https://api.example.com") print(result) asyncio.run(main()) ``` ### Pattern 2: Concurrent Execution with gather() ```python import asyncio from typing import List async def fetch_user(user_id: int) -> dict: """Fetch user data.""" await asyncio.sleep(0.5) return {"id": user_id, "name": f"User {user_id}"} async def fetch_all_users(user_ids: List[int]) -> List[dict]: """Fetch multiple users concurrently.""" tasks = [fetch_user(uid) for uid in user_ids] results = await asyncio.gather(*tasks) return results async def main(): user_ids = [1, 2, 3, 4, 5] users = await fetch_all_users(user_ids) print(f"Fetched {len(users)} users") asyncio.run(main()) ``` ### Pattern 3: Task Creation and Management ```python import asyncio async def background_task(name: str, delay: int): """Long-running background task.""" print(f"{name} started") await asyncio.sleep(delay) print(f"{name} completed") return f"Result from {name}" async def main(): # Create tasks task1 = asyncio.create_task(background_task("Task 1", 2)) task2 = asyncio.create_task(background_task("Task 2", 1)) # Do other work print("Main: doing other work") await asyncio.sleep(0.5) # Wait for tasks result1 = await task1 result2 = await task2 print(f"Results: {result1}, {result2}") asyncio.run(main()) ``` ### Pattern 4: Error Handling in Async Code ```python import asyncio from typing import List, Optional async def risky_operation(item_id: int) -> dict: """Operation that might fail.""" await asyncio.sleep(0.1) if item_id % 3 == 0: raise ValueError(f"Item {item_id} failed") return {"id": item_id, "status": "success"} async def safe_operation(item_id: int) -> Optional[dict]: """Wrapper with error handling.""" try: return await risky_operation(item_id) except ValueError as e: print(f"Error: {e}") return None async def process_items(item_ids: List[int]): """Process multiple items with error handling.""" tasks = [safe_operation(iid) for iid in item_ids] results = await asyncio.gather(*tasks, return_exceptions=True) # Filter out failures successful = [r for r in results if r is not None and not isinstance(r, Exception)] failed = [r for r in results if isinstance(r, Exception)] print(f"Success: {len(successful)}, Failed: {len(failed)}") return successful asyncio.run(process_items([1, 2, 3, 4, 5, 6])) ``` ### Pattern 5: Timeout Handling ```python import asyncio async def slow_operation(delay: int) -> str: """Operation that takes time.""" await asyncio.sleep(delay) return f"Completed after {delay}s" async def with_timeout(): """Execute operation with timeout.""" try: result = await asyncio.wait_for(slow_operation(5), timeout=2.0) print(result) except asyncio.TimeoutError: print("Operation timed out") asyncio.run(with_timeout()) ``` ## Advanced Patterns ### Pattern 6: Async Context Managers ```python import asyncio from typing import Optional class AsyncDatabaseConnection: """Async database connection context manager.""" def __init__(self, dsn: str): self.dsn = dsn self.connection: Optional[object] = None async def __aenter__(self): print("Opening connection") await asyncio.sleep(0.1) # Simulate connection self.connection = {"dsn": self.dsn, "connected": True} return self.connection async def __aexit__(self, exc_type, exc_val, exc_tb): print("Closing connection") await asyncio.sleep(0.1) # Simulate cleanup self.connection = None async def query_database(): """Use async context manager.""" async with AsyncDatabaseConnection("postgresql://localhost") as conn: print(f"Using connection: {conn}") await asyncio.sleep(0.2) # Simulate query return {"rows": 10} asyncio.run(query_database()) ``` ### Pattern 7: Async Iterators and Generators ```python import asyncio from typing import AsyncIterator async def async_range(start: int, end: int, delay: float = 0.1) -> AsyncIterator[int]: """Async generator that yields numbers with delay.""" for i in range(start, end): await asyncio.sleep(delay) yield i async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]: """Fetch paginated data asynchronously.""" for page in range(1, max_pages + 1): await asyncio.sleep(0.2) # Simulate API call yield { "page": page, "url": f"{url}?page={page}", "data": [f"item_{page}_{i}" for i in range(5)] } async def consume_async_iterator(): """Consume async iterator.""" async for number in async_range(1, 5): print(f"Number: {number}") print("\nFetching pages:") async for page_data in fetch_pages("https://api.example.com/items", 3): print(f"Page {page_data['page']}: {len(page_data['data'])} items") asyncio.run(consume_async_iterator()) ``` ### Pattern 8: Producer-Consumer Pattern ```python import asyncio from asyncio import Queue from typing import Optional async def producer(queue: Queue, producer_id: int, num_items: int): """Produce items and put them in queue.""" for i in range(num_items): item = f"Item-{producer_id}-{i}" await queue.put(item) print(f"Producer {producer_id} produced: {item}") await asyncio.sleep(0.1) await queue.put(None) # Signal completion async def consumer(queue: Queue, consumer_id: int): """Consume items from queue.""" while True: item = await queue.get() if item is None: queue.task_done() break print(f"Consumer {consumer_id} processing: {item}") await asyncio.sleep(0.2) # Simulate work queue.task_done() async def producer_consumer_example(): """Run producer-consumer pattern.""" queue = Queue(maxsize=10) # Create tasks producers = [ asyncio.create_task(producer(queue, i, 5)) for i in range(2) ] consumers = [ asyncio.create_task(consumer(queue, i)) for i in range(3) ] # Wait for producers await asyncio.gather(*producers) # Wait for queue to be empty await queue.join() # Cancel consumers for c in consumers: c.cancel() asyncio.run(producer_consumer_example()) ``` ### Pattern 9: Semaphore for Rate Limiting ```python import asyncio from typing import List async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict: """Make API call with rate limiting.""" async with semaphore: print(f"Calling {url}") await asyncio.sleep(0.5) # Simulate API call return {"url": url, "status": 200} async def rate_limited_requests(urls: List[str], max_concurrent: int = 5): """Make multiple requests with rate limiting.""" semaphore = asyncio.Semaphore(max_concurrent) tasks = [api_call(url, semaphore) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [f"https://api.example.com/item/{i}" for i in range(20)] results = await rate_limited_requests(urls, max_concurrent=3) print(f"Completed {len(results)} requests") asyncio.run(main()) ``` ### Pattern 10: Async Locks and Synchronization ```python import asyncio class AsyncCounter: """Thread-safe async counter.""" def __init__(self): self.value = 0 self.lock = asyncio.Lock() async def increment(self): """Safely increment counter.""" async with self.lock: current = self.value await asyncio.sleep(0.01) # Simulate work self.value = current + 1 async def get_value(self) -> int: """Get current value.""" async with self.lock: return self.value async def worker(counter: AsyncCounter, worker_id: int): """Worker that increments counter.""" for _ in range(10): await counter.increment() print(f"Worker {worker_id} incremented") async def test_counter(): """Test concurrent counter.""" counter = AsyncCounter() workers = [asyncio.create_task(worker(counter, i)) for i in range(5)] await asyncio.gather(*workers) final_value = await counter.get_value() print(f"Final counter value: {final_value}") asyncio.run(test_counter()) ``` ## Real-World Applications ### Web Scraping with aiohttp ```python import asyncio import aiohttp from typing import List, Dict async def fetch_url(session: aiohttp.ClientSession, url: str) -> Dict: """Fetch single URL.""" try: async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: text = await response.text() return { "url": url, "status": response.status, "length": len(text) } except Exception as e: return {"url": url, "error": str(e)} async def scrape_urls(urls: List[str]) -> List[Dict]: """Scrape multiple URLs concurrently.""" async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/status/404", ] results = await scrape_urls(urls) for result in results: print(result) asyncio.run(main()) ``` ### Async Database Operations ```python import asyncio from typing import List, Optional # Simulated async database client class AsyncDB: """Simulated async database.""" async def execute(self, query: str) -> List[dict]: """Execute query.""" await asyncio.sleep(0.1) return [{"id": 1, "name": "Example"}] async def fetch_one(self, query: str) -> Optional[dict]: """Fetch single row.""" await asyncio.sleep(0.1) return {"id": 1, "name": "Example"} async def get_user_data(db: AsyncDB, user_id: int) -> dict: """Fetch user and related data concurrently.""" user_task = db.fetch_one(f"SELECT * FROM users WHERE id = {user_id}") orders_task = db.execute(f"SELECT * FROM orders WHERE user_id = {user_id}") profile_task = db.fetch_one(f"SELECT * FROM profiles WHERE user_id = {user_id}") user, orders, profile = await asyncio.gather(user_task, orders_task, profile_task) return { "user": user, "orders": orders, "profile": profile } async def main(): db = AsyncDB() user_data = await get_user_data(db, 1) print(user_data) asyncio.run(main()) ``` ### WebSocket Server ```python import asyncio from typing import Set # Simulated WebSocket connection class WebSocket: """Simulated WebSocket.""" def __init__(self, client_id: str): self.client_id = client_id async def send(self, message: str): """Send message.""" print(f"Sending to {self.client_id}: {message}") await asyncio.sleep(0.01) async def recv(self) -> str: """Receive message.""" await asyncio.sleep(1) return f"Message from {self.client_id}" class WebSocketServer: """Simple WebSocket server.""" def __init__(self): self.clients: Set[WebSocket] = set() async def register(self, websocket: WebSocket): """Register new client.""" self.clients.add(websocket) print(f"Client {websocket.client_id} connected") async def unregister(self, websocket: WebSocket): """Unregister client.""" self.clients.remove(websocket) print(f"Client {websocket.client_id} disconnected") async def broadcast(self, message: str): """Broadcast message to all clients.""" if self.clients: tasks = [client.send(message) for client in self.clients] await asyncio.gather(*tasks) async def handle_client(self, websocket: WebSocket): """Handle individual client connection.""" await self.register(websocket) try: async for message in self.message_iterator(websocket): await self.broadcast(f"{websocket.client_id}: {message}") finally: await self.unregister(websocket) async def message_iterator(self, websocket: WebSocket): """Iterate over messages from client.""" for _ in range(3): # Simulate 3 messages yield await websocket.recv() ``` ## Performance Best Practices ### 1. Use Connection Pools ```python import asyncio import aiohttp async def with_connection_pool(): """Use connection pool for efficiency.""" connector = aiohttp.TCPConnector(limit=100, limit_per_host=10) async with aiohttp.ClientSession(connector=connector) as session: tasks = [session.get(f"https://api.example.com/item/{i}") for i in range(50)] responses = await asyncio.gather(*tasks) return responses ``` ### 2. Batch Operations ```python async def batch_process(items: List[str], batch_size: int = 10): """Process items in batches.""" for i in range(0, len(items), batch_size): batch = items[i:i + batch_size] tasks = [process_item(item) for item in batch] await asyncio.gather(*tasks) print(f"Processed batch {i // batch_size + 1}") async def process_item(item: str): """Process single item.""" await asyncio.sleep(0.1) return f"Processed: {item}" ``` ### 3. Avoid Blocking Operations Never block the event loop with synchronous operations. A single blocking call stalls all concurrent tasks. ```python # BAD - blocks the entire event loop async def fetch_data_bad(): import time import requests time.sleep(1) # Blocks! response = requests.get(url) # Also blocks! # GOOD - use async-native libraries (e.g., httpx for async HTTP) import httpx async def fetch_data_good(url: str): await asyncio.sleep(1) async with httpx.AsyncClient() as client: response = await client.get(url) ``` **Wrapping Blocking Code with `asyncio.to_thread()` (Python 3.9+):** When you must use synchronous libraries, offload to a thread pool: ```python import asyncio from pathlib import Path async def read_file_async(path: str) -> str: """Read file without blocking event loop.""" # asyncio.to_thread() runs sync code in a thread pool return await asyncio.to_thread(Path(path).read_text) async def call_sync_library(data: dict) -> dict: """Wrap a synchronous library call.""" # Useful for sync database drivers, file I/O, CPU work return await asyncio.to_thread(sync_library.process, data) ``` **Lower-level approach with `run_in_executor()`:** ```python import asyncio import concurrent.futures from typing import Any def blocking_operation(data: Any) -> Any: """CPU-intensive blocking operation.""" import time time.sleep(1) return data * 2 async def run_in_executor(data: Any) -> Any: """Run blocking operation in thread pool.""" loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result = await loop.run_in_executor(pool, blocking_operation, data) return result async def main(): results = await asyncio.gather(*[run_in_executor(i) for i in range(5)]) print(results) asyncio.run(main()) ``` ## Common Pitfalls ### 1. Forgetting await ```python # Wrong - returns coroutine object, doesn't execute result = async_function() # Correct result = await async_function() ``` ### 2. Blocking the Event Loop ```python # Wrong - blocks event loop import time async def bad(): time.sleep(1) # Blocks! # Correct async def good(): await asyncio.sleep(1) # Non-blocking ``` ### 3. Not Handling Cancellation ```python async def cancelable_task(): """Task that handles cancellation.""" try: while True: await asyncio.sleep(1) print("Working...") except asyncio.CancelledError: print("Task cancelled, cleaning up...") # Perform cleanup raise # Re-raise to propagate cancellation ``` ### 4. Mixing Sync and Async Code ```python # Wrong - can't call async from sync directly def sync_function(): result = await async_function() # SyntaxError! # Correct def sync_function(): result = asyncio.run(async_function()) ``` ## Testing Async Code ```python import asyncio import pytest # Using pytest-asyncio @pytest.mark.asyncio async def test_async_function(): """Test async function.""" result = await fetch_data("https
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

python-background-jobs

Python background job patterns including task queues, workers, and

coding
⭐1
# Python Background Jobs & Task Queues Decouple long-running or unreliable work from request/response cycles. Return immediately to the user while background workers handle the heavy lifting asynchronously. ## When to Use This Skill - Processing tasks that take longer than a few seconds - Sending emails, notifications, or webhooks - Generating reports or exporting data - Processing uploads or media transformations - Integrating with unreliable external services - Building event-driven architectures ## Core Concepts ### 1. Task Queue Pattern API accepts request, enqueues a job, returns immediately with a job ID. Workers process jobs asynchronously. ### 2. Idempotency Tasks may be retried on failure. Design for safe re-execution. ### 3. Job State Machine Jobs transition through states: pending → running → succeeded/failed. ### 4. At-Least-Once Delivery Most queues guarantee at-least-once delivery. Your code must handle duplicates. ## Quick Start This skill uses Celery for examples, a widely adopted task queue. Alternatives like RQ, Dramatiq, and cloud-native solutions (AWS SQS, GCP Tasks) are equally valid choices. ```python from celery import Celery app = Celery("tasks", broker="redis://localhost:6379") @app.task def send_email(to: str, subject: str, body: str) -> None: # This runs in a background worker email_client.send(to, subject, body) # In your API handler send_email.delay("user@example.com", "Welcome!", "Thanks for signing up") ``` ## Fundamental Patterns ### Pattern 1: Return Job ID Immediately For operations exceeding a few seconds, return a job ID and process asynchronously. ```python from uuid import uuid4 from dataclasses import dataclass from enum import Enum from datetime import datetime class JobStatus(Enum): PENDING = "pending" RUNNING = "running" SUCCEEDED = "succeeded" FAILED = "failed" @dataclass class Job: id: str status: JobStatus created_at: datetime started_at: datetime | None = None completed_at: datetime | None = None result: dict | None = None error: str | None = None # API endpoint async def start_export(request: ExportRequest) -> JobResponse: """Start export job and return job ID.""" job_id = str(uuid4()) # Persist job record await jobs_repo.create(Job( id=job_id, status=JobStatus.PENDING, created_at=datetime.utcnow(), )) # Enqueue task for background processing await task_queue.enqueue( "export_data", job_id=job_id, params=request.model_dump(), ) # Return immediately with job ID return JobResponse( job_id=job_id, status="pending", poll_url=f"/jobs/{job_id}", ) ``` ### Pattern 2: Celery Task Configuration Configure Celery tasks with proper retry and timeout settings. ```python from celery import Celery app = Celery("tasks", broker="redis://localhost:6379") # Global configuration app.conf.update( task_time_limit=3600, # Hard limit: 1 hour task_soft_time_limit=3000, # Soft limit: 50 minutes task_acks_late=True, # Acknowledge after completion task_reject_on_worker_lost=True, worker_prefetch_multiplier=1, # Don't prefetch too many tasks ) @app.task( bind=True, max_retries=3, default_retry_delay=60, autoretry_for=(ConnectionError, TimeoutError), ) def process_payment(self, payment_id: str) -> dict: """Process payment with automatic retry on transient errors.""" try: result = payment_gateway.charge(payment_id) return {"status": "success", "transaction_id": result.id} except PaymentDeclinedError as e: # Don't retry permanent failures return {"status": "declined", "reason": str(e)} except TransientError as e: # Retry with exponential backoff raise self.retry(exc=e, countdown=2 ** self.request.retries * 60) ``` ### Pattern 3: Make Tasks Idempotent Workers may retry on crash or timeout. Design for safe re-execution. ```python @app.task(bind=True) def process_order(self, order_id: str) -> None: """Process order idempotently.""" order = orders_repo.get(order_id) # Already processed? Return early if order.status == OrderStatus.COMPLETED: logger.info("Order already processed", order_id=order_id) return # Already in progress? Check if we should continue if order.status == OrderStatus.PROCESSING: # Use idempotency key to avoid double-charging pass # Process with idempotency key result = payment_provider.charge( amount=order.total, idempotency_key=f"order-{order_id}", # Critical! ) orders_repo.update(order_id, status=OrderStatus.COMPLETED) ``` **Idempotency Strategies:** 1. **Check-before-write**: Verify state before action 2. **Idempotency keys**: Use unique tokens with external services 3. **Upsert patterns**: `INSERT ... ON CONFLICT UPDATE` 4. **Deduplication window**: Track processed IDs for N hours ### Pattern 4: Job State Management Persist job state transitions for visibility and debugging. ```python class JobRepository: """Repository for managing job state.""" async def create(self, job: Job) -> Job: """Create new job record.""" await self._db.execute( """INSERT INTO jobs (id, status, created_at) VALUES ($1, $2, $3)""", job.id, job.status.value, job.created_at, ) return job async def update_status( self, job_id: str, status: JobStatus, **fields, ) -> None: """Update job status with timestamp.""" updates = {"status": status.value, **fields} if status == JobStatus.RUNNING: updates["started_at"] = datetime.utcnow() elif status in (JobStatus.SUCCEEDED, JobStatus.FAILED): updates["completed_at"] = datetime.utcnow() await self._db.execute( "UPDATE jobs SET status = $1, ... WHERE id = $2", updates, job_id, ) logger.info( "Job status updated", job_id=job_id, status=status.value, ) ``` ## Advanced Patterns ### Pattern 5: Dead Letter Queue Handle permanently failed tasks for manual inspection. ```python @app.task(bind=True, max_retries=3) def process_webhook(self, webhook_id: str, payload: dict) -> None: """Process webhook with DLQ for failures.""" try: result = send_webhook(payload) if not result.success: raise WebhookFailedError(result.error) except Exception as e: if self.request.retries >= self.max_retries: # Move to dead letter queue for manual inspection dead_letter_queue.send({ "task": "process_webhook", "webhook_id": webhook_id, "payload": payload, "error": str(e), "attempts": self.request.retries + 1, "failed_at": datetime.utcnow().isoformat(), }) logger.error( "Webhook moved to DLQ after max retries", webhook_id=webhook_id, error=str(e), ) return # Exponential backoff retry raise self.retry(exc=e, countdown=2 ** self.request.retries * 60) ``` ### Pattern 6: Status Polling Endpoint Provide an endpoint for clients to check job status. ```python from fastapi import FastAPI, HTTPException app = FastAPI() @app.get("/jobs/{job_id}") async def get_job_status(job_id: str) -> JobStatusResponse: """Get current status of a background job.""" job = await jobs_repo.get(job_id) if job is None: raise HTTPException(404, f"Job {job_id} not found") return JobStatusResponse( job_id=job.id, status=job.status.value, created_at=job.created_at, started_at=job.started_at, completed_at=job.completed_at, result=job.result if job.status == JobStatus.SUCCEEDED else None, error=job.error if job.status == JobStatus.FAILED else None, # Helpful for clients is_terminal=job.status in (JobStatus.SUCCEEDED, JobStatus.FAILED), ) ``` ### Pattern 7: Task Chaining and Workflows Compose complex workflows from simple tasks. ```python from celery import chain, group, chord # Simple chain: A → B → C workflow = chain( extract_data.s(source_id), transform_data.s(), load_data.s(destination_id), ) # Parallel execution: A, B, C all at once parallel = group( send_email.s(user_email), send_sms.s(user_phone), update_analytics.s(event_data), ) # Chord: Run tasks in parallel, then a callback # Process all items, then send completion notification workflow = chord( [process_item.s(item_id) for item_id in item_ids], send_completion_notification.s(batch_id), ) workflow.apply_async() ``` ### Pattern 8: Alternative Task Queues Choose the right tool for your needs. **RQ (Redis Queue)**: Simple, Redis-based ```python from rq import Queue from redis import Redis queue = Queue(connection=Redis()) job = queue.enqueue(send_email, "user@example.com", "Subject", "Body") ``` **Dramatiq**: Modern Celery alternative ```python import dramatiq from dramatiq.brokers.redis import RedisBroker dramatiq.set_broker(RedisBroker()) @dramatiq.actor def send_email(to: str, subject: str, body: str) -> None: email_client.send(to, subject, body) ``` **Cloud-native options:** - AWS SQS + Lambda - Google Cloud Tasks - Azure Functions ## Best Practices Summary 1. **Return immediately** - Don't block requests for long operations 2. **Persist job state** - Enable status polling and debugging 3. **Make tasks idempotent** - Safe to retry on any failure 4. **Use idempotency keys** - For external service calls 5. **Set timeouts** - Both soft and hard limits 6. **Implement DLQ** - Capture permanently failed tasks 7. **Log transitions** - Track job state changes 8. **Retry appropriately** - Exponential backoff for transient errors 9. **Don't retry permanent failures** - Validation errors, invalid credentials 10. **Monitor queue depth** - Alert on backlog growth
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

python-observability

Python observability patterns including structured logging,

coding
⭐1
# Python Observability Instrument Python applications with structured logs, metrics, and traces. When something breaks in production, you need to answer "what, where, and why" without deploying new code. ## When to Use This Skill - Adding structured logging to applications - Implementing metrics collection with Prometheus - Setting up distributed tracing across services - Propagating correlation IDs through request chains - Debugging production issues - Building observability dashboards ## Core Concepts ### 1. Structured Logging Emit logs as JSON with consistent fields for production environments. Machine-readable logs enable powerful queries and alerts. For local development, consider human-readable formats. ### 2. The Four Golden Signals Track latency, traffic, errors, and saturation for every service boundary. ### 3. Correlation IDs Thread a unique ID through all logs and spans for a single request, enabling end-to-end tracing. ### 4. Bounded Cardinality Keep metric label values bounded. Unbounded labels (like user IDs) explode storage costs. ## Quick Start ```python import structlog structlog.configure( processors=[ structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer(), ], ) logger = structlog.get_logger() logger.info("Request processed", user_id="123", duration_ms=45) ``` ## Fundamental Patterns ### Pattern 1: Structured Logging with Structlog Configure structlog for JSON output with consistent fields. ```python import logging import structlog def configure_logging(log_level: str = "INFO") -> None: """Configure structured logging for the application.""" structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.JSONRenderer(), ], wrapper_class=structlog.make_filtering_bound_logger( getattr(logging, log_level.upper()) ), context_class=dict, logger_factory=structlog.PrintLoggerFactory(), cache_logger_on_first_use=True, ) # Initialize at application startup configure_logging("INFO") logger = structlog.get_logger() ``` ### Pattern 2: Consistent Log Fields Every log entry should include standard fields for filtering and correlation. ```python import structlog from contextvars import ContextVar # Store correlation ID in context correlation_id: ContextVar[str] = ContextVar("correlation_id", default="") logger = structlog.get_logger() def process_request(request: Request) -> Response: """Process request with structured logging.""" logger.info( "Request received", correlation_id=correlation_id.get(), method=request.method, path=request.path, user_id=request.user_id, ) try: result = handle_request(request) logger.info( "Request completed", correlation_id=correlation_id.get(), status_code=200, duration_ms=elapsed, ) return result except Exception as e: logger.error( "Request failed", correlation_id=correlation_id.get(), error_type=type(e).__name__, error_message=str(e), ) raise ``` ### Pattern 3: Semantic Log Levels Use log levels consistently across the application. | Level | Purpose | Examples | |-------|---------|----------| | `DEBUG` | Development diagnostics | Variable values, internal state | | `INFO` | Request lifecycle, operations | Request start/end, job completion | | `WARNING` | Recoverable anomalies | Retry attempts, fallback used | | `ERROR` | Failures needing attention | Exceptions, service unavailable | ```python # DEBUG: Detailed internal information logger.debug("Cache lookup", key=cache_key, hit=cache_hit) # INFO: Normal operational events logger.info("Order created", order_id=order.id, total=order.total) # WARNING: Abnormal but handled situations logger.warning( "Rate limit approaching", current_rate=950, limit=1000, reset_seconds=30, ) # ERROR: Failures requiring investigation logger.error( "Payment processing failed", order_id=order.id, error=str(e), payment_provider="stripe", ) ``` Never log expected behavior at `ERROR`. A user entering a wrong password is `INFO`, not `ERROR`. ### Pattern 4: Correlation ID Propagation Generate a unique ID at ingress and thread it through all operations. ```python from contextvars import ContextVar import uuid import structlog correlation_id: ContextVar[str] = ContextVar("correlation_id", default="") def set_correlation_id(cid: str | None = None) -> str: """Set correlation ID for current context.""" cid = cid or str(uuid.uuid4()) correlation_id.set(cid) structlog.contextvars.bind_contextvars(correlation_id=cid) return cid # FastAPI middleware example from fastapi import Request async def correlation_middleware(request: Request, call_next): """Middleware to set and propagate correlation ID.""" # Use incoming header or generate new cid = request.headers.get("X-Correlation-ID") or str(uuid.uuid4()) set_correlation_id(cid) response = await call_next(request) response.headers["X-Correlation-ID"] = cid return response ``` Propagate to outbound requests: ```python import httpx async def call_downstream_service(endpoint: str, data: dict) -> dict: """Call downstream service with correlation ID.""" async with httpx.AsyncClient() as client: response = await client.post( endpoint, json=data, headers={"X-Correlation-ID": correlation_id.get()}, ) return response.json() ``` ## Advanced Patterns ### Pattern 5: The Four Golden Signals with Prometheus Track these metrics for every service boundary: ```python from prometheus_client import Counter, Histogram, Gauge # Latency: How long requests take REQUEST_LATENCY = Histogram( "http_request_duration_seconds", "Request latency in seconds", ["method", "endpoint", "status"], buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10], ) # Traffic: Request rate REQUEST_COUNT = Counter( "http_requests_total", "Total HTTP requests", ["method", "endpoint", "status"], ) # Errors: Error rate ERROR_COUNT = Counter( "http_errors_total", "Total HTTP errors", ["method", "endpoint", "error_type"], ) # Saturation: Resource utilization DB_POOL_USAGE = Gauge( "db_connection_pool_used", "Number of database connections in use", ) ``` Instrument your endpoints: ```python import time from functools import wraps def track_request(func): """Decorator to track request metrics.""" @wraps(func) async def wrapper(request: Request, *args, **kwargs): method = request.method endpoint = request.url.path start = time.perf_counter() try: response = await func(request, *args, **kwargs) status = str(response.status_code) return response except Exception as e: status = "500" ERROR_COUNT.labels( method=method, endpoint=endpoint, error_type=type(e).__name__, ).inc() raise finally: duration = time.perf_counter() - start REQUEST_COUNT.labels(method=method, endpoint=endpoint, status=status).inc() REQUEST_LATENCY.labels(method=method, endpoint=endpoint, status=status).observe(duration) return wrapper ``` ### Pattern 6: Bounded Cardinality Avoid labels with unbounded values to prevent metric explosion. ```python # BAD: User ID has potentially millions of values REQUEST_COUNT.labels(method="GET", user_id=user.id) # Don't do this! # GOOD: Bounded values only REQUEST_COUNT.labels(method="GET", endpoint="/users", status="200") # If you need per-user metrics, use a different approach: # - Log the user_id and query logs # - Use a separate analytics system # - Bucket users by type/tier REQUEST_COUNT.labels( method="GET", endpoint="/users", user_tier="premium", # Bounded set of values ) ``` ### Pattern 7: Timed Operations with Context Manager Create a reusable timing context manager for operations. ```python from contextlib import contextmanager import time import structlog logger = structlog.get_logger() @contextmanager def timed_operation(name: str, **extra_fields): """Context manager for timing and logging operations.""" start = time.perf_counter() logger.debug("Operation started", operation=name, **extra_fields) try: yield except Exception as e: elapsed_ms = (time.perf_counter() - start) * 1000 logger.error( "Operation failed", operation=name, duration_ms=round(elapsed_ms, 2), error=str(e), **extra_fields, ) raise else: elapsed_ms = (time.perf_counter() - start) * 1000 logger.info( "Operation completed", operation=name, duration_ms=round(elapsed_ms, 2), **extra_fields, ) # Usage with timed_operation("fetch_user_orders", user_id=user.id): orders = await order_repository.get_by_user(user.id) ``` ### Pattern 8: OpenTelemetry Tracing Set up distributed tracing with OpenTelemetry. **Note:** OpenTelemetry is actively evolving. Check the [official Python documentation](https://opentelemetry.io/docs/languages/python/) for the latest API patterns and best practices. ```python from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter def configure_tracing(service_name: str, otlp_endpoint: str) -> None: """Configure OpenTelemetry tracing.""" provider = TracerProvider() processor = BatchSpanProcessor(OTLPSpanExporter(endpoint=otlp_endpoint)) provider.add_span_processor(processor) trace.set_tracer_provider(provider) tracer = trace.get_tracer(__name__) async def process_order(order_id: str) -> Order: """Process order with tracing.""" with tracer.start_as_current_span("process_order") as span: span.set_attribute("order.id", order_id) with tracer.start_as_current_span("validate_order"): validate_order(order_id) with tracer.start_as_current_span("charge_payment"): charge_payment(order_id) with tracer.start_as_current_span("send_confirmation"): send_confirmation(order_id) return order ``` ## Best Practices Summary 1. **Use structured logging** - JSON logs with consistent fields 2. **Propagate correlation IDs** - Thread through all requests and logs 3. **Track the four golden signals** - Latency, traffic, errors, saturation 4. **Bound label cardinality** - Never use unbounded values as metric labels 5. **Log at appropriate levels** - Don't cry wolf with ERROR 6. **Include context** - User ID, request ID, operation name in logs 7. **Use context managers** - Consistent timing and error handling 8. **Separate concerns** - Observability code shouldn't pollute business logic 9. **Test your observability** - Verify logs and metrics in integration tests 10. **Set up alerts** - Metrics are useless without alerting
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

uv-package-manager

Master the uv package manager for fast Python dependency

coding
⭐1
# UV Package Manager Comprehensive guide to using uv, an extremely fast Python package installer and resolver written in Rust, for modern Python project management and dependency workflows. ## When to Use This Skill - Setting up new Python projects quickly - Managing Python dependencies faster than pip - Creating and managing virtual environments - Installing Python interpreters - Resolving dependency conflicts efficiently - Migrating from pip/pip-tools/poetry - Speeding up CI/CD pipelines - Managing monorepo Python projects - Working with lockfiles for reproducible builds - Optimizing Docker builds with Python dependencies ## Core Concepts ### 1. What is uv? - **Ultra-fast package installer**: 10-100x faster than pip - **Written in Rust**: Leverages Rust's performance - **Drop-in pip replacement**: Compatible with pip workflows - **Virtual environment manager**: Create and manage venvs - **Python installer**: Download and manage Python versions - **Resolver**: Advanced dependency resolution - **Lockfile support**: Reproducible installations ### 2. Key Features - Blazing fast installation speeds - Disk space efficient with global cache - Compatible with pip, pip-tools, poetry - Comprehensive dependency resolution - Cross-platform support (Linux, macOS, Windows) - No Python required for installation - Built-in virtual environment support ### 3. UV vs Traditional Tools - **vs pip**: 10-100x faster, better resolver - **vs pip-tools**: Faster, simpler, better UX - **vs poetry**: Faster, less opinionated, lighter - **vs conda**: Faster, Python-focused ## Installation ### Quick Install ```bash # macOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # Windows (PowerShell) powershell -c "irm https://astral.sh/uv/install.ps1 | iex" # Using pip (if you already have Python) pip install uv # Using Homebrew (macOS) brew install uv # Using cargo (if you have Rust) cargo install --git https://github.com/astral-sh/uv uv ``` ### Verify Installation ```bash uv --version # uv 0.x.x ``` ## Quick Start ### Create a New Project ```bash # Create new project with virtual environment uv init my-project cd my-project # Or create in current directory uv init . # Initialize creates: # - .python-version (Python version) # - pyproject.toml (project config) # - README.md # - .gitignore ``` ### Install Dependencies ```bash # Install packages (creates venv if needed) uv add requests pandas # Install dev dependencies uv add --dev pytest black ruff # Install from requirements.txt uv pip install -r requirements.txt # Install from pyproject.toml uv sync ``` ## Virtual Environment Management ### Pattern 1: Creating Virtual Environments ```bash # Create virtual environment with uv uv venv # Create with specific Python version uv venv --python 3.12 # Create with custom name uv venv my-env # Create with system site packages uv venv --system-site-packages # Specify location uv venv /path/to/venv ``` ### Pattern 2: Activating Virtual Environments ```bash # Linux/macOS source .venv/bin/activate # Windows (Command Prompt) .venv\Scripts\activate.bat # Windows (PowerShell) .venv\Scripts\Activate.ps1 # Or use uv run (no activation needed) uv run python script.py uv run pytest ``` ### Pattern 3: Using uv run ```bash # Run Python script (auto-activates venv) uv run python app.py # Run installed CLI tool uv run black . uv run pytest # Run with specific Python version uv run --python 3.11 python script.py # Pass arguments uv run python script.py --arg value ``` ## Package Management ### Pattern 4: Adding Dependencies ```bash # Add package (adds to pyproject.toml) uv add requests # Add with version constraint uv add "django>=4.0,<5.0" # Add multiple packages uv add numpy pandas matplotlib # Add dev dependency uv add --dev pytest pytest-cov # Add optional dependency group uv add --optional docs sphinx # Add from git uv add git+https://github.com/user/repo.git # Add from git with specific ref uv add git+https://github.com/user/repo.git@v1.0.0 # Add from local path uv add ./local-package # Add editable local package uv add -e ./local-package ``` ### Pattern 5: Removing Dependencies ```bash # Remove package uv remove requests # Remove dev dependency uv remove --dev pytest # Remove multiple packages uv remove numpy pandas matplotlib ``` ### Pattern 6: Upgrading Dependencies ```bash # Upgrade specific package uv add --upgrade requests # Upgrade all packages uv sync --upgrade # Upgrade package to latest uv add --upgrade requests # Show what would be upgraded uv tree --outdated ``` ### Pattern 7: Locking Dependencies ```bash # Generate uv.lock file uv lock # Update lock file uv lock --upgrade # Lock without installing uv lock --no-install # Lock specific package uv lock --upgrade-package requests ``` ## Python Version Management ### Pattern 8: Installing Python Versions ```bash # Install Python version uv python install 3.12 # Install multiple versions uv python install 3.11 3.12 3.13 # Install latest version uv python install # List installed versions uv python list # Find available versions uv python list --all-versions ``` ### Pattern 9: Setting Python Version ```bash # Set Python version for project uv python pin 3.12 # This creates/updates .python-version file # Use specific Python version for command uv --python 3.11 run python script.py # Create venv with specific version uv venv --python 3.12 ``` ## Project Configuration ### Pattern 10: pyproject.toml with uv ```toml [project] name = "my-project" version = "0.1.0" description = "My awesome project" readme = "README.md" requires-python = ">=3.8" dependencies = [ "requests>=2.31.0", "pydantic>=2.0.0", "click>=8.1.0", ] [project.optional-dependencies] dev = [ "pytest>=7.4.0", "pytest-cov>=4.1.0", "black>=23.0.0", "ruff>=0.1.0", "mypy>=1.5.0", ] docs = [ "sphinx>=7.0.0", "sphinx-rtd-theme>=1.3.0", ] [build-system] requires = ["hatchling"] build-backend = "hatchling.build" [tool.uv] dev-dependencies = [ # Additional dev dependencies managed by uv ] [tool.uv.sources] # Custom package sources my-package = { git = "https://github.com/user/repo.git" } ``` ### Pattern 11: Using uv with Existing Projects ```bash # Migrate from requirements.txt uv add -r requirements.txt # Migrate from poetry # Already have pyproject.toml, just use: uv sync # Export to requirements.txt uv pip freeze > requirements.txt # Export with hashes uv pip freeze --require-hashes > requirements.txt ``` ## Advanced Workflows ### Pattern 12: Monorepo Support ```bash # Project structure # monorepo/ # packages/ # package-a/ # pyproject.toml # package-b/ # pyproject.toml # pyproject.toml (root) # Root pyproject.toml [tool.uv.workspace] members = ["packages/*"] # Install all workspace packages uv sync # Add workspace dependency uv add --path ./packages/package-a ``` ### Pattern 13: CI/CD Integration ```yaml # .github/workflows/test.yml name: Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Install uv uses: astral-sh/setup-uv@v2 with: enable-cache: true - name: Set up Python run: uv python install 3.12 - name: Install dependencies run: uv sync --all-extras --dev - name: Run tests run: uv run pytest - name: Run linting run: | uv run ruff check . uv run black --check . ``` ### Pattern 14: Docker Integration ```dockerfile # Dockerfile FROM python:3.12-slim # Install uv COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv # Set working directory WORKDIR /app # Copy dependency files COPY pyproject.toml uv.lock ./ # Install dependencies RUN uv sync --frozen --no-dev # Copy application code COPY . . # Run application CMD ["uv", "run", "python", "app.py"] ``` **Optimized multi-stage build:** ```dockerfile # Multi-stage Dockerfile FROM python:3.12-slim AS builder # Install uv COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv WORKDIR /app # Install dependencies to venv COPY pyproject.toml uv.lock ./ RUN uv sync --frozen --no-dev --no-editable # Runtime stage FROM python:3.12-slim WORKDIR /app # Copy venv from builder COPY --from=builder /app/.venv .venv COPY . . # Use venv ENV PATH="/app/.venv/bin:$PATH" CMD ["python", "app.py"] ``` ### Pattern 15: Lockfile Workflows ```bash # Create lockfile (uv.lock) uv lock # Install from lockfile (exact versions) uv sync --frozen # Update lockfile without installing uv lock --no-install # Upgrade specific package in lock uv lock --upgrade-package requests # Check if lockfile is up to date uv lock --check # Export lockfile to requirements.txt uv export --format requirements-txt > requirements.txt # Export with hashes for security uv export --format requirements-txt --hash > requirements.txt ``` ## Performance Optimization ### Pattern 16: Using Global Cache ```bash # UV automatically uses global cache at: # Linux: ~/.cache/uv # macOS: ~/Library/Caches/uv # Windows: %LOCALAPPDATA%\uv\cache # Clear cache uv cache clean # Check cache size uv cache dir ``` ### Pattern 17: Parallel Installation ```bash # UV installs packages in parallel by default # Control parallelism uv pip install --jobs 4 package1 package2 # No parallel (sequential) uv pip install --jobs 1 package ``` ### Pattern 18: Offline Mode ```bash # Install from cache only (no network) uv pip install --offline package # Sync from lockfile offline uv sync --frozen --offline ``` ## Comparison with Other Tools ### uv vs pip ```bash # pip python -m venv .venv source .venv/bin/activate pip install requests pandas numpy # ~30 seconds # uv uv venv uv add requests pandas numpy # ~2 seconds (10-15x faster) ``` ### uv vs poetry ```bash # poetry poetry init poetry add requests pandas poetry install # ~20 seconds # uv uv init uv add requests pandas uv sync # ~3 seconds (6-7x faster) ``` ### uv vs pip-tools ```bash # pip-tools pip-compile requirements.in pip-sync requirements.txt # ~15 seconds # uv uv lock uv sync --frozen # ~2 seconds (7-8x faster) ``` ## Common Workflows ### Pattern 19: Starting a New Project ```bash # Complete workflow uv init my-project cd my-project # Set Python version uv python pin 3.12 # Add dependencies uv add fastapi uvicorn pydantic # Add dev dependencies uv add --dev pytest black ruff mypy # Create structure mkdir -p src/my_project tests # Run tests uv run pytest # Format code uv run black . uv run ruff check . ``` ### Pattern 20: Maintaining Existing Project ```bash # Clone repository git clone https://github.com/user/project.git cd project # Install dependencies (creates venv automatically) uv sync # Install with dev dependencies uv sync --all-extras # Update dependencies uv lock --upgrade # Run application uv run python app.py # Run tests uv run pytest # Add new dependency uv add new-package # Commit updated files git add pyproject.toml uv.lock git commit -m "Add new-package dependency" ``` ## Tool Integration ### Pattern 21: Pre-commit Hooks ```yaml # .pre-commit-config.yaml repos: - repo: local hooks: - id: uv-lock name: uv lock entry: uv lock language: system pass_filenames: false - id: ruff name: ruff entry: uv run ruff check --fix language: system types: [python] - id: black name: black entry: uv run black language: system types: [python] ``` ### Pattern 22: VS Code Integration ```json // .vscode/settings.json { "python.defaultInterpreterPath": "${workspaceFolder}/.venv/bin/python", "python.terminal.activateEnvironment": true, "python.testing.pytestEnabled": true, "python.testing.pytestArgs": ["-v"], "python.linting.enabled": true, "python.formatting.provider": "black", "[python]": { "editor.defaultFormatter": "ms-python.black-formatter", "editor.formatOnSave": true } } ``` ## Troubleshooting ### Common Issues ```bash # Issue: uv not found # Solution: Add to PATH or reinstall echo 'export PATH="$HOME/.cargo/bin:$PATH"' >> ~/.bashrc # Issue: Wrong Python version # Solution: Pin version explicitly uv python pin 3.12 uv venv --python 3.12 # Issue: Dependency conflict # Solution: Check resolution uv lock --verbose # Issue: Cache issues # Solution: Clear cache uv cache clean # Issue: Lockfile out of sync # Solution: Regenerate uv lock --upgrade ``` ## Best Practices ### Project Setup 1. **Always use lockfiles** for reproducibility 2. **Pin Python version** with .python-version 3. **Separate dev dependencies** from production 4. **Use uv run** instead of activating venv 5. **Commit uv.lock** to version control 6. **Use --frozen in CI** for consistent builds 7. **Leverage global cache** for speed 8. **Use workspace** for monorepos 9. **Export requirements.txt** for compatibility 10. **Keep uv updated** for latest features ### Performance Tips ```bash # Use frozen installs in CI uv sync --frozen # Use offline mode when possible uv sync --offline # Parallel operations (automatic) # uv does this by default # Reuse cache across environments # uv shares cache globally # Use lockfiles to skip resolution uv sync --frozen # skips resolution ``` ## Migration Guide ### From pip + requirements.txt ```bash # Before python -m venv .venv source .venv/bin/activate pip install -r requirements.txt # After uv venv uv pip install -r requirements.txt # Or better: uv init uv add -r requirements.txt ``` ### From Poetry ```bash # Before poetry install poetry add requests # After uv sync uv add requests # Keep existing pyproject.toml # uv reads [project] and [tool.poetry] sections ``` ### From pip-tools ```bash # Before pip-compile requirements.in pip-sync requirements.txt # After uv lock uv sync --frozen ``` ## Command Reference ### Essential Commands ```bash # Project management uv init [PATH] # Initialize project uv add PACKAGE # Add dependency uv remove PACKAGE # Remove dependency uv sync # Install dependencies uv lock # Create/update lockfile # Virtual environments uv venv [PATH] # Create venv uv run COMMAND # Run in venv # Python management uv python install VERSION # Install Python uv python list # List installed Pythons uv python pin VERSION # Pin Python version # Package installation (pip-compatible) uv pip install PACKAGE # Install package uv pip uninstall PACKAGE # Uninstall package uv pip freeze # List installed uv pip list # List packages # Utility uv cache clean # Clear cache uv cache dir # Show cache location uv --version # Show version ``` ## Resources - **Official documentation**: https://docs.astral.sh/uv/ - **GitHub repository**: https://github.com/astral-sh/uv - **Astral blog**: https://astral.sh/blog - **Migration guides**: https://docs.astral.sh/uv/guides/ - **Comparison with other tools**: https://docs.astral.sh/uv/pip/compatibility/ ## Best Practices Summary 1. **Use uv for all new projects** - Start with `uv init` 2. **Commit lockfiles** - Ensure reproducible builds 3. **Pin Python versions** - Use .python-version 4. **Use uv run** - Avoid manual venv activation 5. **Leverage caching** - Let uv manage global cache 6. **Use --frozen in CI** - Exact reproduction 7. **Keep uv updated** - Fast-moving project 8. **Use workspaces** - For monorepo projects 9. **Export for compatibility** - Generate requirements.txt when needed 10. **Read the docs** - uv is feature-rich and evolving
👍0
👁️0
🤖 Auto-discovered