Python is the glue language of the industry — from web APIs to ML pipelines to automation scripts. These recipes produce typed, async-first Python code that uses modern patterns: Pydantic for validation, SQLAlchemy 2.0 for databases, and pytest for testing. No more untyped dictionaries passed between functions.
FastAPI and Django REST framework recipes with proper typing
Database integration with SQLAlchemy 2.0 async and Alembic migrations
Background task patterns with Celery and async queues
Testing recipes with pytest, fixtures, and mocking
Scenario: You need a new API service that is testable, typed, and follows Python best practices.
Create a FastAPI project with proper structure. Directories: app/api/ for routers, app/core/ for config and security, app/models/ for SQLAlchemy models, app/schemas/ for Pydantic models, app/services/ for business logic, app/repositories/ for data access. Set up: (1) app/core/config.py using pydantic-settings to load and validate all environment variables at startup with typed defaults. (2) app/core/deps.py with FastAPI dependencies: get_db() yields an async SQLAlchemy session, get_current_user() decodes JWT and returns User, get_current_admin() checks role. (3) Health check at GET /health. (4) CORS middleware configured from environment variables. (5) Structured logging with structlog, request ID middleware. (6) Exception handlers for 422 (Pydantic), 404, 401, 500. (7) Run with uvicorn and gunicorn for production. Write tests using pytest-asyncio and httpx AsyncClient.
Expected output: Project structure, config, dependencies, middleware, exception handlers, and tests.
Scenario: Your API accepts nested JSON with cross-field validation rules that Pydantic v2 can handle but you keep writing raw dictionaries.
Create Pydantic v2 schemas for a product catalog API in app/schemas/product.py. (1) ProductCreate: name (str, 2-200 chars), description (str, optional, max 5000), price (Decimal, positive, max 2 decimal places via a custom validator), compareAtPrice (optional Decimal, must be greater than price if present — cross-field validation), categoryId (UUID), tags (list of str, max 10, each max 50 chars), variants (list of ProductVariantCreate, min 1), metadata (dict with string keys, max 20 entries). (2) ProductVariantCreate: sku (str matching regex ^[A-Z0-9-]+$), size (enum), color (str), stock (non-negative int), weight (optional positive float). (3) ProductResponse: includes all create fields plus id, slug (auto-generated from name), createdAt, updatedAt, and nested variants with their IDs. (4) ProductListResponse: paginated with items, total, page, pageSize, hasMore. Use model_config with from_attributes = True for ORM compatibility. Test validation with valid data, every invalid field case, and cross-field validation failure.
Expected output: Pydantic schemas with validators, response models, and comprehensive validation tests.
Scenario: Your database code uses raw SQL strings with no type safety, no migration support, and synchronous calls blocking the event loop.
Set up SQLAlchemy 2.0 async with PostgreSQL. (1) Create app/models/base.py with a declarative base class that includes id (UUID, default uuid4), created_at, and updated_at columns on every model. (2) Create models: User (email unique, name, hashed_password, role, is_active), Product (name, slug unique, description, price Decimal(10,2), stock, category relationship), Category (name unique, slug, parent_id self-referential for hierarchy), Order (user relationship, status enum, total, items relationship). (3) Create app/core/database.py with async engine using asyncpg, async session factory, and a session dependency. Connection pool: pool_size=10, max_overflow=20, pool_timeout=30. (4) Create app/repositories/product.py with typed async methods: get_by_id(id), list(filters, page, page_size) with dynamic filtering and sorting, create(data), update(id, data), soft_delete(id). Use select() with joinedload for eager loading relationships. (5) Set up Alembic for async migrations with autogenerate. Create initial migration. Test queries using an async SQLite in-memory database with pytest fixtures.
Expected output: Base model, 4 domain models, async database setup, repository pattern, Alembic config, and tests.
Scenario: Your API generates PDF reports synchronously, taking 30 seconds per request. Users are frustrated.
Set up Celery with Redis broker for background task processing. (1) app/worker/celery_app.py: configure Celery with Redis broker and result backend, task serialization with JSON, task time limits (soft 5 min, hard 10 min), and automatic retry on connection failure. (2) Create tasks: generate_report(user_id, report_type, params) — queries database, generates PDF with ReportLab, uploads to S3, sends email notification with download link. send_transactional_email(to, template, context) — renders Jinja2 template, sends via SMTP/Resend. process_csv_import(file_path, user_id) — reads CSV in chunks of 1000 rows, validates each row, bulk-inserts valid rows, returns summary of successes and failures. (3) Create a task status API: POST /api/tasks/{task_id}/status that returns Celery task state and result. (4) Add Celery Beat for scheduled tasks: daily cleanup of expired sessions, weekly analytics aggregation. (5) Add Flower for monitoring at /admin/tasks. Test tasks using Celery’s eager mode (CELERY_ALWAYS_EAGER=True) in tests.
Expected output: Celery config, 3 task definitions, status API, Beat schedule, and eager-mode tests.
Scenario: Your FastAPI project has zero tests. You need to test routes, services, and database interactions.
Create a comprehensive test setup for FastAPI with pytest. (1) tests/conftest.py: create an async test database (SQLite in-memory), override the database dependency, create an httpx AsyncClient fixture, create a fixture that creates and authenticates a test user, create a factory fixture for generating test data using factory_boy. (2) tests/api/test_products.py: test CRUD endpoints — create returns 201 with correct fields, list supports pagination and filtering, get by ID returns 200 or 404, update returns 200 with changed fields only, delete returns 204 and subsequent get returns 404. Test auth: unauthenticated request returns 401, non-admin creating product returns 403. Test validation: every invalid field returns 422 with specific error message. (3) tests/services/test_product_service.py: unit tests with mocked repository. (4) tests/repositories/test_product_repo.py: integration tests with real async database. (5) conftest.py fixtures should clean the database between tests using transactions that roll back. Add pytest-cov with 80% minimum coverage.
Expected output: conftest with fixtures, API tests, service tests, repository tests, and coverage config.
Scenario: You are building a multi-tenant SaaS with Django and need role-based access control on every endpoint.
Create a Django REST Framework API for a project management app. (1) Models: Project (name, description, owner FK to User, created_at), Task (title, description, status enum, priority enum, project FK, assignee FK, due_date), Comment (content, task FK, author FK, created_at). (2) Serializers: nested serializers for reading (ProjectSerializer includes task count and owner name), flat serializers for writing (ProjectCreateSerializer with just name and description). (3) ViewSets: ProjectViewSet with custom permissions — owner can CRUD, team members can read and update, others cannot access. TaskViewSet filtered by project, with status transition validation (cannot go from done back to todo). (4) Custom permissions: IsProjectOwner, IsProjectMember, IsTaskAssignee. (5) Filtering with django-filter: tasks filterable by status, priority, assignee, due date range. (6) Pagination: cursor-based pagination for tasks (better for real-time updates than offset). (7) Multi-tenancy: override get_queryset on every viewset to filter by the current user’s organization. Test every permission scenario and filter combination.
Expected output: Models, serializers, viewsets, permissions, filters, and permission tests.
Scenario: Your service calls three external APIs and cascading failures bring everything down.
Create a resilient HTTP client using httpx with retry and circuit breaker patterns. (1) app/clients/base.py: create a BaseHttpClient class with httpx.AsyncClient, configurable timeout (connect: 5s, read: 30s), retry logic (3 attempts with exponential backoff for 5xx and connection errors, no retry for 4xx), and structured logging of every request/response. (2) Implement a circuit breaker: track failure rate over a sliding window (last 10 requests). If failure rate exceeds 50%, open the circuit for 30 seconds (all requests fail immediately with CircuitOpenError). After 30 seconds, allow one probe request — if it succeeds, close the circuit; if it fails, reset the timer. (3) Create typed clients: StripeClient (create charge, create customer, list invoices), SendGridClient (send email, send template), SlackClient (post message, post to channel). Each method returns a typed response or raises a typed exception. (4) Health check integration: the /health endpoint reports the circuit state of each external dependency. Test the retry logic, circuit breaker state transitions, and timeout handling using respx for mocking.
Expected output: Base client with retry/circuit breaker, 3 typed API clients, health integration, and tests.
Scenario: Your team runs manual database scripts and deployment tasks. You need a typed CLI tool.
Build a CLI management tool using Typer. (1) cli/main.py with command groups: db (migrate, seed, reset, backup), users (create-admin, reset-password, list, deactivate), tasks (run-now, list-scheduled, clear-queue), deploy (check-health, rollback, scale). (2) db migrate runs Alembic migrations and reports which migrations were applied. db seed generates test data with a —count flag. db reset drops and recreates with confirmation prompt. db backup creates a pg_dump and uploads to S3. (3) users create-admin prompts for email and password interactively with validation. (4) Add --format option (table, json, csv) for all list commands using rich for table formatting. (5) Add --dry-run flag for destructive operations. (6) Create a @requires_config decorator that loads and validates the config file before running the command. (7) Add shell completion generation. Test commands using Typer’s testing utilities with CliRunner.
Expected output: CLI app with 4 command groups, formatted output, dry-run support, and CliRunner tests.
Scenario: You need a real-time chat feature in your FastAPI application.
Implement WebSocket chat with FastAPI. (1) WebSocket endpoint at /ws/chat/{room_id} that authenticates via a token query parameter, joins the user to a room, and broadcasts messages. (2) Create a ConnectionManager class: track active connections per room in a dict, handle connect (authenticate, add to room, broadcast join notification), handle disconnect (remove from room, broadcast leave), handle message (validate, store in database, broadcast to room). (3) Message types: text, image (URL), system (join/leave/typing). Store all messages in a messages table with room_id, user_id, type, content, created_at. (4) Load last 50 messages on connect for history. (5) Typing indicators: client sends ‘typing’ event, server broadcasts to room (debounced). (6) For horizontal scaling, use Redis pub/sub: each server instance subscribes to room channels, messages published to Redis are received by all instances. (7) Rate limit messages: max 10 per second per user. Test using the websocket test client from httpx.
Expected output: WebSocket endpoint, connection manager, Redis pub/sub, message persistence, and tests.
Scenario: You need to process a 10 GB CSV file without loading it into memory.
Build a streaming data pipeline using async generators. (1) Create app/pipeline/reader.py with an async generator that reads a CSV file in chunks using aiofiles, yielding parsed rows one at a time. Handle encoding detection and BOM stripping. (2) Create app/pipeline/transformer.py with async generators for each transformation step: validate_rows(source) yields valid rows and logs invalid ones with line numbers, enrich_rows(source) adds computed fields (geocoding addresses, looking up external IDs), batch_rows(source, size=1000) groups rows into batches for bulk operations. (3) Create app/pipeline/loader.py that bulk-inserts batches into PostgreSQL using COPY for maximum throughput, handling conflicts with ON CONFLICT DO UPDATE. (4) Compose the pipeline: reader | validate | enrich | batch | load. Report progress: rows processed, rows skipped, elapsed time, estimated time remaining. (5) Add checkpointing: save progress to a state file every 10,000 rows so the pipeline can resume after a crash. Test the full pipeline with a small CSV and verify row counts at each stage.
Expected output: Reader, transformer, loader modules, pipeline composition, checkpointing, and tests.
Scenario: Your app reads environment variables with os.getenv scattered everywhere, no validation, and defaults that differ between files.
Create a centralized configuration system using pydantic-settings. (1) app/core/config.py with a Settings class inheriting from BaseSettings: DATABASE_URL (PostgresDsn, required), REDIS_URL (RedisDsn, default localhost), SECRET_KEY (SecretStr, required, min 32 chars), DEBUG (bool, default False), ALLOWED_ORIGINS (list of str, comma-separated in env), LOG_LEVEL (Literal[‘DEBUG’, ‘INFO’, ‘WARNING’, ‘ERROR’], default INFO), AWS_ACCESS_KEY (SecretStr, optional), AWS_SECRET_KEY (SecretStr, optional), SMTP_HOST (str, optional), environment-specific: ENV (Literal[‘development’, ‘staging’, ‘production’]). (2) Add computed properties: is_production, database_url_async (converts sync to async URL). (3) Load from .env file in development, from environment in production. (4) Create a get_settings() function with lru_cache for singleton access. (5) Create a app/core/config_test.py with test-specific settings that override database URL to SQLite. (6) Validate that all required variables are set at startup — fail fast with clear error messages listing every missing variable. Test with various env combinations and verify validation errors.
Expected output: Settings class, cached getter, test overrides, startup validation, and tests.
Scenario: “It works on my machine” is your team’s most-used phrase. Everyone has different Python versions and system dependencies.
Create Docker configuration for development and production. (1) Dockerfile: multi-stage build — builder stage installs Poetry and dependencies, production stage copies only the virtual environment and application code. Use python:3.12-slim, create non-root user, set PYTHONUNBUFFERED=1 and PYTHONDONTWRITEBYTECODE=1, health check using curl to /health. (2) docker-compose.yml for development: app with volume mount for hot reload (use watchfiles), PostgreSQL 16 with initialization script, Redis 7, Mailhog for email testing, pgAdmin for database management. (3) docker-compose.prod.yml: app with gunicorn + uvicorn workers (calculated from CPU count), PostgreSQL with persistent volume and backup cron job, Redis with password and persistence, Nginx as reverse proxy with SSL termination. (4) .dockerignore excluding tests, docs, .git, pycache , .env. (5) Makefile with shortcuts: make dev, make test, make build, make deploy, make migrate, make shell. Test the build completes and health check passes.
Expected output: Multi-stage Dockerfile, compose files, .dockerignore, Makefile, and build tests.
Common Python pitfalls:
Async/sync mixing: If the AI calls a synchronous function (like open() or synchronous ORM queries) inside an async handler, it blocks the event loop. Ensure all I/O uses async variants.Pydantic v1 vs v2: AI tools sometimes generate Pydantic v1 syntax (class Config, validator decorator). If you use Pydantic v2, check for model_config dict and field_validator decorator.Import cycles: The AI may create circular imports between models and schemas. Use TYPE_CHECKING imports or reorganize dependencies.SQLAlchemy session leaks: If the AI forgets to close sessions in error paths, connections leak. Always use the dependency injection pattern with yield and finally.
