Python to język-klej branży — od API webowych przez potoki ML po skrypty automatyzacji. Te przepisy produkują typowany, asynchroniczny w pierwszej kolejności kod Python, który używa nowoczesnych wzorców: Pydantic do walidacji, SQLAlchemy 2.0 do baz danych i pytest do testowania. Koniec z nietypowanymi słownikami przekazywanymi między funkcjami.
Przepisy na FastAPI i Django REST Framework z prawidłowym typowaniem
Integracja z bazą danych przez SQLAlchemy 2.0 async i migracje Alembic
Wzorce zadań w tle z Celery i kolejkami asynchronicznymi
Przepisy na testy z pytest, fixtures i mockingiem
Te prompty są niezależne od narzędzia: wklej je identycznie do trybu agenta Cursor, Claude Code (claude "<prompt>" lub interaktywnie) albo Codex (codex "<prompt>", rozszerzenie IDE czy zadanie Cloud na worktree). Wygenerowany kod i sposób, w jaki go oceniasz, są takie same — zmienia się tylko powierzchnia, na której piszesz. Poniższe notki MCP/skills pokazują, gdzie podłączone narzędzie zacieśnia pętlę.
Scenariusz: Potrzebujesz nowego serwisu API, który jest testowalny, typowany i zgodny z najlepszymi praktykami Pythona.
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.
Oczekiwany wynik: Struktura projektu, konfiguracja, zależności, middleware, handlery wyjątków i testy.
Scenariusz: Twoje API akceptuje zagnieżdżony JSON z regułami walidacji między polami, które Pydantic v2 potrafi obsłużyć, ale ty ciągle piszesz surowe słowniki.
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.
Oczekiwany wynik: Schematy Pydantic z walidatorami, modele odpowiedzi i kompleksowe testy walidacji.
Scenariusz: Twój kod bazodanowy używa surowych ciągów SQL bez bezpieczeństwa typów, bez wsparcia migracji i z synchronicznymi wywołaniami blokującymi pętlę zdarzeń.
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.
Oczekiwany wynik: Model bazowy, 4 modele domenowe, konfiguracja asynchronicznej bazy, wzorzec repozytorium, konfiguracja Alembic i testy.
Podłącz serwer Postgres MCP podczas budowania tego przepisu, żeby agent czytał twój żywy schemat zamiast wymyślać nazwy kolumn: claude mcp add --transport stdio postgres -- npx -y @modelcontextprotocol/server-postgres "$DATABASE_URL" (ten sam pakiet podpięty przez .mcp.json w Cursor i Codex; pakiet jest zarchiwizowany u źródła, ale wciąż się instaluje i działa). Różnica przed/po jest konkretna — “add a joinedload for the category relationship” staje się potwierdzeniem przez agenta prawdziwego klucza obcego i indeksu przed napisaniem zapytania, zamiast poprawiania zhalucynowanego category_fk po fakcie. W developmencie wskaż mu rolę tylko do odczytu.
Scenariusz: Twoje API generuje raporty PDF synchronicznie, co zajmuje 30 sekund na żądanie. Użytkownicy są sfrustrowani.
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).
Oczekiwany wynik: Konfiguracja Celery, 3 definicje zadań, API statusu, harmonogram Beat i testy w trybie eager.
Scenariusz: Twój projekt FastAPI ma zero testów. Musisz przetestować trasy, serwisy i interakcje z bazą danych.
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.
Oczekiwany wynik: conftest z fixtures, testy API, testy serwisów, testy repozytorium i konfiguracja pokrycia.
Scenariusz: Budujesz wielodostępny SaaS z Django i potrzebujesz kontroli dostępu opartej na rolach na każdym endpoincie.
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.
Oczekiwany wynik: Modele, serializery, viewsety, uprawnienia, filtry i testy uprawnień.
Scenariusz: Twój serwis wywołuje trzy zewnętrzne API i kaskadowe awarie zawalają wszystko.
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.
Oczekiwany wynik: Bazowy klient z retry/circuit breaker, 3 typowane klienty API, integracja health i testy.
Budujesz StripeClient? Dodaj zdalny serwer MCP Stripe’a (claude mcp add --transport http stripe https://mcp.stripe.com; ten sam URL w Cursor i Codex przez .mcp.json), żeby agent mógł sprawdzić dokładne aktualne kształty API — pola charge vs. PaymentIntent, najnowsze kody błędów — prosto od Stripe’a, a nie z nieaktualnych danych treningowych. To różnica między klientem, który kompiluje się względem dzisiejszego API, a takim, który odwołuje się do parametru usuniętego dwie wersje temu. Do jednorazowej generacji bez trwałego połączenia lżejszą opcją jest skill: przejrzyj skills.sh i zainstaluj jednozadaniowy skill przez npx skills add <owner/repo> (CLI vercel-labs/skills; działa w Claude Code, Cursor i Codex). Zasada kciuka: skill do jednorazowego wzmocnienia, serwer MCP, gdy agent musi nieustannie odpytywać Stripe’a.
Scenariusz: Twój zespół uruchamia ręczne skrypty bazodanowe i zadania wdrożeniowe. Potrzebujesz typowanego narzędzia CLI.
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.
Oczekiwany wynik: Aplikacja CLI z 4 grupami poleceń, formatowanym wyjściem, wsparciem dry-run i testami CliRunner.
Scenariusz: Potrzebujesz funkcji czatu w czasie rzeczywistym w swojej aplikacji FastAPI.
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.
Oczekiwany wynik: Endpoint WebSocket, menedżer połączeń, Redis pub/sub, trwałość wiadomości i testy.
Scenariusz: Musisz przetworzyć plik CSV o rozmiarze 10 GB bez ładowania go do pamięci.
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.
Oczekiwany wynik: Moduły reader, transformer, loader, kompozycja potoku, punkty kontrolne i testy.
Scenariusz: Twoja aplikacja odczytuje zmienne środowiskowe z os.getenv rozsianymi wszędzie, bez walidacji i z wartościami domyślnymi różniącymi się między plikami.
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.
Oczekiwany wynik: Klasa Settings, cached getter, nadpisania testowe, walidacja przy starcie i testy.
Scenariusz: “Na moim komputerze działa” to najczęściej używane zdanie w twoim zespole. Każdy ma inne wersje Pythona i zależności systemowe.
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.13-slim (current stable; python:3.12-slim is also fine if a dependency lags), 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.
Oczekiwany wynik: Wieloetapowy Dockerfile, pliki compose, .dockerignore, Makefile i testy builda.
Typowe pułapki Pythona:
Mieszanie async/sync: Jeśli AI wywołuje synchroniczną funkcję (jak open() lub synchroniczne zapytania ORM) wewnątrz asynchronicznego handlera, blokuje to pętlę zdarzeń. Upewnij się, że całe I/O używa wariantów async.Pydantic v1 vs v2: Narzędzia AI czasem generują składnię Pydantic v1 (class Config, dekorator validator). Jeśli używasz Pydantic v2, sprawdzaj słownik model_config i dekorator field_validator.Cykliczne importy: AI może tworzyć cykliczne importy między modelami i schematami. Użyj importów TYPE_CHECKING lub zreorganizuj zależności.Wycieki sesji SQLAlchemy: Jeśli AI zapomni zamknąć sesję w ścieżkach błędów, połączenia wyciekają. Zawsze używaj wzorca wstrzykiwania zależności z yield i finally.
