Python to język-klej branży — od API webowych przez potoki ML po skrypty automatyzacji. Te przepisy produkują typowany, asynchroniczny kod Python używający 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
Scenariusz: Potrzebujesz nowego serwisu API, który jest testowalny, typowany i zgodny z najlepszymi praktykami Pythona.
Stwórz projekt FastAPI z prawidłową strukturą. Katalogi: app/api/ dla routerów, app/core/ dla konfiguracji i bezpieczeństwa, app/models/ dla modeli SQLAlchemy, app/schemas/ dla modeli Pydantic, app/services/ dla logiki biznesowej, app/repositories/ dla dostępu do danych. Skonfiguruj: (1) app/core/config.py używając pydantic-settings do ładowania i walidacji wszystkich zmiennych środowiskowych przy starcie z typowanymi wartościami domyślnymi. (2) app/core/deps.py z zależnościami FastAPI: get_db() zwraca asynchroniczną sesję SQLAlchemy, get_current_user() dekoduje JWT i zwraca User, get_current_admin() sprawdza rolę. (3) Health check na GET /health. (4) Middleware CORS konfigurowany ze zmiennych środowiskowych. (5) Strukturalne logowanie z structlog, middleware request ID. (6) Handlery wyjątków dla 422 (Pydantic), 404, 401, 500. (7) Uruchamianie z uvicorn i gunicorn do produkcji. Napisz testy używając pytest-asyncio i httpx AsyncClient.
Oczekiwany wynik: Struktura projektu, konfiguracja, zależności, middleware, handlery wyjątków i testy.
Scenariusz: Twoje API akceptuje zagnieżdżone JSON z regułami walidacji między polami, które Pydantic v2 potrafi obsługiwać, ale ty ciągle piszesz surowe słowniki.
Stwórz schematy Pydantic v2 dla API katalogu produktów w app/schemas/product.py. (1) ProductCreate: name (str, 2-200 znaków), description (str, opcjonalny, max 5000), price (Decimal, dodatni, max 2 miejsca dziesiętne przez niestandardowy walidator), compareAtPrice (opcjonalny Decimal, musi być większy niż price jeśli obecny — walidacja między polami), categoryId (UUID), tags (lista str, max 10, każdy max 50 znaków), variants (lista ProductVariantCreate, min 1), metadata (słownik z kluczami string, max 20 wpisów). (2) ProductVariantCreate: sku (str pasujący do regex ^[A-Z0-9-]+$), size (enum), color (str), stock (nieujemny int), weight (opcjonalny dodatni float). (3) ProductResponse: zawiera wszystkie pola create plus id, slug (automatycznie generowany z name), createdAt, updatedAt i zagnieżdżone warianty z ich ID. (4) ProductListResponse: paginowany z items, total, page, pageSize, hasMore. Użyj model_config z from_attributes = True dla kompatybilności z ORM. Testuj walidacje z prawidłowymi danymi, każdym nieprawidłowym polem i awarią walidacji między polami.
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 synchronicznych wywołań blokujących pętlę zdarzeń.
Skonfiguruj SQLAlchemy 2.0 async z PostgreSQL. (1) Stwórz app/models/base.py z deklaratywną klasą bazową, która zawiera kolumny id (UUID, default uuid4), created_at i updated_at w każdym modelu. (2) Stwórz modele: User (email unique, name, hashed_password, role, is_active), Product (name, slug unique, description, price Decimal(10,2), stock, relacja category), Category (name unique, slug, parent_id samoreferencyjna dla hierarchii), Order (relacja user, status enum, total, relacja items). (3) Stwórz app/core/database.py z asynchronicznym silnikiem używając asyncpg, fabryka sesji async i zależność sesji. Pula połączeń: pool_size=10, max_overflow=20, pool_timeout=30. (4) Stwórz app/repositories/product.py z typowanymi metodami async: get_by_id(id), list(filters, page, page_size) z dynamicznym filtrowaniem i sortowaniem, create(data), update(id, data), soft_delete(id). Użyj select() z joinedload do zachłannego ładowania relacji. (5) Skonfiguruj Alembic dla migracji async z autogeneracją. Stwórz początkową migrację. Testuj zapytania używając asynchronicznej bazy SQLite w pamięci z fixtures pytest.
Oczekiwany wynik: Model bazowy, 4 modele domenowe, konfiguracja asynchronicznej bazy, wzorzec repozytorium, konfiguracja Alembic i testy.
Scenariusz: Twoje API generuje raporty PDF synchronicznie, co zajmuje 30 sekund na żądanie. Użytkownicy są sfrustrowani.
Skonfiguruj Celery z brokerem Redis do przetwarzania zadań w tle. (1) app/worker/celery_app.py: skonfiguruj Celery z brokerem i backendem wyników Redis, serializacja zadań z JSON, limity czasu (soft 5 min, hard 10 min) i automatyczne ponawianie przy awarii połączenia. (2) Stwórz zadania: generate_report(user_id, report_type, params) — odpytuje bazę danych, generuje PDF z ReportLab, przesyła do S3, wysyła powiadomienie e-mail z linkiem do pobrania. send_transactional_email(to, template, context) — renderuje szablon Jinja2, wysyła przez SMTP/Resend. process_csv_import(file_path, user_id) — czyta CSV w porcjach po 1000 wierszy, waliduje każdy wiersz, wstawia masowo prawidłowe wiersze, zwraca podsumowanie sukcesów i porażek. (3) Stwórz API statusu zadań: POST /api/tasks/{task_id}/status zwracające stan i wynik zadania Celery. (4) Dodaj Celery Beat do zaplanowanych zadań: codzienne czyszczenie wygasłych sesji, tygodniowa agregacja analityk. (5) Dodaj Flower do monitorowania pod /admin/tasks. Testuj zadania używając trybu eager Celery (CELERY_ALWAYS_EAGER=True) w testach.
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.
Stwórz kompleksową konfigurację testów dla FastAPI z pytest. (1) tests/conftest.py: stwórz asynchroniczną testową bazę danych (SQLite w pamięci), nadpisz zależność bazy danych, stwórz fixture httpx AsyncClient, stwórz fixture tworzący i uwierzytelniający testowego użytkownika, stwórz fixture fabryki do generowania danych testowych używając factory_boy. (2) tests/api/test_products.py: testuj endpointy CRUD — create zwraca 201 z prawidłowymi polami, list wspiera paginację i filtrowanie, get by ID zwraca 200 lub 404, update zwraca 200 tylko ze zmienionymi polami, delete zwraca 204 i następny get zwraca 404. Testuj auth: nieuwierzytelnione żądanie zwraca 401, nie-admin tworzący produkt zwraca 403. Testuj walidację: każde nieprawidłowe pole zwraca 422 z konkretnym komunikatem błędu. (3) tests/services/test_product_service.py: testy jednostkowe z zamockowanym repozytorium. (4) tests/repositories/test_product_repo.py: testy integracyjne z prawdziwą asynchroniczną bazą danych. (5) Fixtures w conftest.py powinny czyścić bazę między testami używając transakcji, które są wycofywane. Dodaj pytest-cov z minimalnym pokryciem 80%.
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.
Stwórz API Django REST Framework dla aplikacji do zarządzania projektami. (1) Modele: Project (name, description, owner FK do 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) Serializery: zagnieżdżone serializery do odczytu (ProjectSerializer zawiera liczbę zadań i nazwę właściciela), płaskie serializery do zapisu (ProjectCreateSerializer tylko z name i description). (3) ViewSets: ProjectViewSet z niestandardowymi uprawnieniami — właściciel może CRUD, członkowie zespołu mogą czytać i aktualizować, inni nie mają dostępu. TaskViewSet filtrowany po projekcie, z walidacją przejść statusu (nie można wrócić z done do todo). (4) Niestandardowe uprawnienia: IsProjectOwner, IsProjectMember, IsTaskAssignee. (5) Filtrowanie z django-filter: zadania filtrowalne po status, priority, assignee, zakresie due date. (6) Paginacja: paginacja oparta na kursorze dla zadań (lepsza dla aktualizacji w czasie rzeczywistym niż offset). (7) Wielodostępność: nadpisz get_queryset w każdym viewsecie aby filtrować po organizacji bieżącego użytkownika. Testuj każdy scenariusz uprawnień i kombinacje filtrów.
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.
Stwórz odporny klient HTTP używając httpx z wzorcami retry i circuit breaker. (1) app/clients/base.py: stwórz klasę BaseHttpClient z httpx.AsyncClient, konfigurowalnym timeout (connect: 5s, read: 30s), logiką ponawiania (3 próby z wykładniczym opóźnieniem dla 5xx i błędów połączenia, brak ponawiania dla 4xx) i strukturalnym logowaniem każdego żądania/odpowiedzi. (2) Zaimplementuj circuit breaker: śledź wskaźnik błędów w przesuwnym oknie (ostatnie 10 żądań). Jeśli wskaźnik błędów przekroczy 50%, otwórz obwód na 30 sekund (wszystkie żądania natychmiast kończą się CircuitOpenError). Po 30 sekundach, pozwól na jedno żądanie sondujące — jeśli się powiedzie, zamknij obwód; jeśli zawiedzie, zresetuj timer. (3) Stwórz typowane klienty: StripeClient (create charge, create customer, list invoices), SendGridClient (send email, send template), SlackClient (post message, post to channel). Każda metoda zwraca typowaną odpowiedź lub rzuca typowany wyjątek. (4) Integracja z health check: endpoint /health raportuje stan obwodu każdej zewnętrznej zależności. Testuj logikę ponawiania, przejścia stanów circuit breaker i obsługę timeoutów używając respx do mockowania.
Oczekiwany wynik: Bazowy klient z retry/circuit breaker, 3 typowane klienty API, integracja health i testy.
Scenariusz: Twój zespół uruchamia ręczne skrypty bazodanowe i zadania wdrożeniowe. Potrzebujesz typowanego narzędzia CLI.
Zbuduj narzędzie zarządzania CLI używając Typer. (1) cli/main.py z grupami poleceń: 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 uruchamia migracje Alembic i raportuje które migracje zostały zastosowane. db seed generuje dane testowe z flagą —count. db reset usuwa i odtwarza z potwierdzeniem. db backup tworzy pg_dump i przesyła do S3. (3) users create-admin pyta interaktywnie o email i hasło z walidacją. (4) Dodaj opcję --format (table, json, csv) dla wszystkich poleceń listowania używając rich do formatowania tabel. (5) Dodaj flagę --dry-run dla destrukcyjnych operacji. (6) Stwórz dekorator @requires_config, który ładuje i waliduje plik konfiguracyjny przed uruchomieniem polecenia. (7) Dodaj generowanie podpowiedzi powłoki. Testuj polecenia używając narzędzi testowych Typer z 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.
Zaimplementuj czat WebSocket z FastAPI. (1) Endpoint WebSocket pod /ws/chat/{room_id}, który uwierzytelnia przez parametr zapytania token, dołącza użytkownika do pokoju i rozgłasza wiadomości. (2) Stwórz klasę ConnectionManager: śledź aktywne połączenia per pokój w słowniku, obsługuj connect (uwierzytelnij, dodaj do pokoju, rozgłoś powiadomienie o dołączeniu), obsługuj disconnect (usuń z pokoju, rozgłoś wyjście), obsługuj message (waliduj, zapisz w bazie danych, rozgłoś do pokoju). (3) Typy wiadomości: text, image (URL), system (join/leave/typing). Przechowuj wszystkie wiadomości w tabeli messages z room_id, user_id, type, content, created_at. (4) Ładuj ostatnie 50 wiadomości przy połączeniu dla historii. (5) Wskaźniki pisania: klient wysyła zdarzenie ‘typing’, serwer rozgłasza do pokoju (z debouncingiem). (6) Dla horyzontalnego skalowania użyj Redis pub/sub: każda instancja serwera subskrybuje kanały pokojów, wiadomości publikowane do Redis są odbierane przez wszystkie instancje. (7) Ograniczanie liczby wiadomości: max 10 na sekundę na użytkownika. Testuj używając klienta testowego WebSocket z 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.
Zbuduj strumieniowy potok danych używając generatorów asynchronicznych. (1) Stwórz app/pipeline/reader.py z generatorem asynchronicznym, który czyta plik CSV w porcjach używając aiofiles, zwracając sparsowane wiersze po jednym. Obsługuj detekcję kodowania i usuwanie BOM. (2) Stwórz app/pipeline/transformer.py z generatorami asynchronicznymi dla każdego kroku transformacji: validate_rows(source) zwraca prawidłowe wiersze i loguje nieprawidłowe z numerami linii, enrich_rows(source) dodaje obliczone pola (geokodowanie adresów, wyszukiwanie zewnętrznych ID), batch_rows(source, size=1000) grupuje wiersze w partie do operacji masowych. (3) Stwórz app/pipeline/loader.py, który masowo wstawia partie do PostgreSQL używając COPY dla maksymalnej przepustowości, obsługując konflikty z ON CONFLICT DO UPDATE. (4) Skomponuj potok: reader | validate | enrich | batch | load. Raportuj postęp: przetworzone wiersze, pominięte wiersze, czas trwania, szacowany czas do zakończenia. (5) Dodaj punkty kontrolne: zapisuj postęp do pliku stanu co 10 000 wierszy, aby potok mógł wznowić pracę po awarii. Testuj pełny potok z małym plikiem CSV i weryfikuj liczbę wierszy na każdym etapie.
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.
Stwórz scentralizowany system konfiguracji używając pydantic-settings. (1) app/core/config.py z klasą Settings dziedziczącą po BaseSettings: DATABASE_URL (PostgresDsn, wymagany), REDIS_URL (RedisDsn, domyślnie localhost), SECRET_KEY (SecretStr, wymagany, min 32 znaki), DEBUG (bool, domyślnie False), ALLOWED_ORIGINS (lista str, rozdzielone przecinkami w env), LOG_LEVEL (Literal[‘DEBUG’, ‘INFO’, ‘WARNING’, ‘ERROR’], domyślnie INFO), AWS_ACCESS_KEY (SecretStr, opcjonalny), AWS_SECRET_KEY (SecretStr, opcjonalny), SMTP_HOST (str, opcjonalny), specyficzne dla środowiska: ENV (Literal[‘development’, ‘staging’, ‘production’]). (2) Dodaj właściwości obliczane: is_production, database_url_async (konwertuje synchroniczny na asynchroniczny URL). (3) Ładuj z pliku .env w developmencie, ze środowiska w produkcji. (4) Stwórz funkcję get_settings() z lru_cache dla dostępu singleton. (5) Stwórz app/core/config_test.py z ustawieniami specyficznymi dla testów, które nadpisują URL bazy danych na SQLite. (6) Waliduj, że wszystkie wymagane zmienne są ustawione przy starcie — szybko zawiedź z czytelnymi komunikatami błędów wymieniającymi każdą brakującą zmienną. Testuj z różnymi kombinacjami env i weryfikuj błędy walidacji.
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.
Stwórz konfigurację Docker dla developmentu i produkcji. (1) Dockerfile: wieloetapowy build — etap builder instaluje Poetry i zależności, etap produkcyjny kopiuje tylko środowisko wirtualne i kod aplikacji. Użyj python:3.12-slim, stwórz użytkownika nie-root, ustaw PYTHONUNBUFFERED=1 i PYTHONDONTWRITEBYTECODE=1, health check używając curl do /health. (2) docker-compose.yml dla developmentu: aplikacja z zamontowanym wolumenem dla hot reload (użyj watchfiles), PostgreSQL 16 ze skryptem inicjalizacyjnym, Redis 7, Mailhog do testowania e-maili, pgAdmin do zarządzania bazą. (3) docker-compose.prod.yml: aplikacja z gunicorn + workerami uvicorn (obliczanymi z liczby CPU), PostgreSQL z trwałym wolumenem i zadaniem cron do backupów, Redis z hasłem i trwałością, Nginx jako reverse proxy z terminacją SSL. (4) .dockerignore wykluczający tests, docs, .git, pycache , .env. (5) Makefile ze skrótami: make dev, make test, make build, make deploy, make migrate, make shell. Testuj, czy build kończy się sukcesem i health check przechodzi.
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.
