🐍 Python — FastAPI & ETL

Работа с Python проектами через Claude Code: FastAPI, SQLAlchemy 2, Alembic, Pydantic v2, Celery. Стандарты кода (Ruff + mypy), ETL-паттерны и pytest-asyncio.

⚙️ Стандарты кода Python

Ruff — замена flake8 + black + isort

# pyproject.toml
[tool.ruff]
line-length = 100
target-version = "py312"
src = ["app", "tests"]

[tool.ruff.lint]
select = [
  "E",   # pycodestyle errors
  "W",   # pycodestyle warnings
  "F",   # pyflakes
  "I",   # isort
  "N",   # pep8-naming
  "UP",  # pyupgrade (modern Python syntax)
  "B",   # flake8-bugbear
  "SIM", # flake8-simplify
  "RUF", # ruff-specific
]
ignore = ["E501"]  # line too long handled by formatter

[tool.ruff.lint.isort]
known-first-party = ["app"]

[tool.ruff.format]
quote-style = "double"
indent-style = "space"

mypy — строгая типизация

# pyproject.toml
[tool.mypy]
python_version = "3.12"
strict = true             # Все строгие проверки включены
ignore_missing_imports = true

# Запуск:
# mypy app/
# или через docker MCP:
# mcp__docker__docker_exec({ container: "etl-app-1", command: "mypy app/" })

💡

auto-format hook автоматически запускает ruff check --fix && ruff format после каждого изменения .py файла. Ручной запуск не нужен.

⚡ FastAPI — паттерны

Структура эндпоинта

from __future__ import annotations

from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from app.deps import get_db, get_current_user
from app.schemas import PaymentCreate, PaymentResponse
from app.services import PaymentService

router = APIRouter(prefix="/payments", tags=["payments"])

# Annotated для DI — правило проекта
CurrentUser = Annotated[User, Depends(get_current_user)]
DbSession = Annotated[AsyncSession, Depends(get_db)]

@router.post("/", response_model=PaymentResponse, status_code=status.HTTP_201_CREATED)
async def create_payment(
    data: PaymentCreate,          # Pydantic model — не dict!
    user: CurrentUser,
    db: DbSession,
) -> PaymentResponse:
    try:
        return await PaymentService(db).create(data, user_id=user.id)
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            detail=str(e)  # Понятное сообщение — правило проекта
        ) from e

Pydantic v2 схемы

from pydantic import BaseModel, Field, ConfigDict
from decimal import Decimal

class PaymentCreate(BaseModel):
    amount: Decimal = Field(gt=0, description="Amount in cents")
    currency: str = Field(min_length=3, max_length=3, pattern=r"^[A-Z]{3}$")

class PaymentResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # Pydantic v2!

    id: int
    amount: Decimal
    currency: str
    status: str

🔄 ETL — паттерны безопасности

from __future__ import annotations

import logging
from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger(__name__)

BATCH_SIZE = 10_000  # Правило: не более 10K строк за транзакцию

async def etl_load_payments(session: AsyncSession, records: list[dict]) -> int:
    """ETL шаг: загрузка платежей."""
    logger.info(f"ETL step=load_payments start count={len(records)}")

    loaded = 0
    for i in range(0, len(records), BATCH_SIZE):
        batch = records[i : i + BATCH_SIZE]
        try:
            async with session.begin():  # Атомарная транзакция
                session.add_all([Payment(**r) for r in batch])
            loaded += len(batch)
        except Exception as e:
            logger.error(f"ETL step=load_payments batch={i} error={e}")
            await session.rollback()          # Откат при ошибке
            await record_etl_error(session, step="load_payments", error=str(e))
            raise  # Пробрасываем — except: pass запрещён!

    logger.info(f"ETL step=load_payments end loaded={loaded}")
    return loaded

🚫

except Exception: pass — запрещён! Это скрывает ошибки и делает отладку невозможной. Правило: всегда логировать (logger.error) или пробрасывать (raise).

🧪 pytest-asyncio — тесты

# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"  # Все async тесты автоматически

# tests/test_payments.py
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine

# Тестовая БД — ВСЕГДА SQLite in-memory (никакого prod коннекта!)
TEST_DB = "sqlite+aiosqlite:///:memory:"

@pytest.fixture
async def db_session():
    engine = create_async_engine(TEST_DB)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with AsyncSession(engine) as session:
        yield session

@pytest.mark.asyncio
async def test_create_payment(db_session, client: AsyncClient):
    # Arrange
    payload = {"amount": "1000.00", "currency": "RUB"}

    # Act
    response = await client.post("/api/v1/payments", json=payload)

    # Assert
    assert response.status_code == 201
    assert response.json()["amount"] == "1000.00"

🔧 Celery — асинхронные задачи

from celery import Celery
import os

# Настройка через os.getenv — никаких секретов в коде!
celery_app = Celery(
    "etl_tasks",
    broker=os.getenv("REDIS_URL", "redis://redis:6379/0"),
    backend=os.getenv("REDIS_URL", "redis://redis:6379/0"),
)

@celery_app.task(bind=True, max_retries=3)
def process_payment_batch(self, batch_id: int) -> dict:
    try:
        return _do_processing(batch_id)
    except Exception as exc:
        # Retry с exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

📏 Ключевые правила Python проектов

Правило	Плохо	Хорошо
Импорты аннотаций	`def f(x: list[str]): ...` (Python <3.9)	`from __future__ import annotations` вверху файла
env переменные	`os.environ["SECRET_KEY"]`	`settings = Settings()` через pydantic BaseSettings
Исключения	`except Exception: pass`	`except Exception as e: logger.error(e); raise`
Запросы к БД	Строчная конкатенация SQL	SQLAlchemy ORM или параметризованные запросы
Тестовая БД	Подключение к реальной БД	SQLite in-memory `sqlite+aiosqlite:///:memory:`
Type hints	`def f(data):`	`def f(data: dict[str, Any]) -> PaymentResponse:`