Python backend patterns for AI coding assistants span building production backends with FastAPI and Django, adding typed Pydantic validation, SQLAlchemy 2.0 database access, Celery background tasks, and pytest test suites. Each recipe pairs a concrete prompt with the async-first code it produces, so Cursor, Claude Code, and Codex generate consistent, testable Python across routing, background work, and Docker deployment.
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
These prompts are tool-agnostic: paste them identically into Cursor agent mode, Claude Code (claude "<prompt>" or interactively), or Codex (codex "<prompt>", the IDE extension, or a Cloud task on a worktree). The generated code and the way you evaluate it are the same — only the surface where you type changes. The MCP/skills notes below show where a connected tool tightens the loop.
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.
If an approved database connector is available, point it at a read-only development role so the agent reads the real schema instead of inventing columns. Connector packages and configuration differ by host; verify the maintained server and the host’s MCP instructions before adding one. Without a connector, export the relevant schema into repository context and require the agent to cite the real foreign keys and indexes before writing queries.
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 — in Celery 5.x use the lowercase setting task_always_eager = True (the legacy uppercase CELERY_ALWAYS_EAGER still works via the old settings namespace, but prefer the modern lowercase keys).
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.
Building the StripeClient? Add Stripe’s remote MCP server (claude mcp add --transport http stripe https://mcp.stripe.com; same URL in Cursor and Codex via .mcp.json) so the agent can look up the exact current API shapes — charge vs. PaymentIntent fields, the latest error codes — straight from Stripe instead of from stale training data. That is the difference between a client that compiles against today’s API and one that references a parameter removed two versions ago. For one-shot generation without a persistent connection, a skill is the lighter option: browse skills.sh and install a single-purpose skill with npx skills add <owner/repo> (the vercel-labs/skills CLI; works across Claude Code, Cursor, and Codex). Rule of thumb: skill for single-purpose augmentation, MCP server when the agent needs to keep querying Stripe.
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) Read pyproject.toml, lock files, and deployment constraints; use the supported Python major already declared by the project, pin the exact image digest, and do not infer the latest release. Install dependencies in a builder stage, copy only runtime files into a non-root stage, set Python runtime environment variables, and add a health check. (2) Create a local Compose stack for the application and its development dependencies. (3) Keep production databases and caches outside the application container, define resource limits and secret injection, and use the serving topology supported by the chosen framework version. (4) Exclude credentials, caches, and development outputs in .dockerignore. (5) Test the image build, startup, signal handling, health check, and rollback. Confirm compatibility against Python’s official release pages .
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.