orderflow/
├── pyproject.toml
└── src/orderflow/
    ├── domain/          # сущности и правила. Импортирует ТОЛЬКО stdlib
    ├── app/             # сценарии использования и порты (Protocol). Импортирует domain
    ├── adapters/        # postgres/, payments/, events/. Импортируют app и domain
    ├── entrypoints/     # api/, worker/, cli/. Импортируют всё, кроме друг друга
    ├── config.py        # Settings — читается один раз
    ├── logging_setup.py # dictConfig — вызывается один раз
    └── bootstrap.py     # точка сборки: создаёт адаптеры и связывает

# src/orderflow/domain/order.py — ни одного импорта из adapters, фреймворков и драйверов
from __future__ import annotations

from dataclasses import dataclass, replace
from decimal import Decimal
from enum import StrEnum


class OrderStatus(StrEnum):
    NEW = "new"
    PAID = "paid"
    CANCELLED = "cancelled"


class OrderError(Exception):
    """База для ошибок предметной области: транспорт ловит её, а не голый Exception."""


class AlreadyPaid(OrderError):
    pass


class CannotCancelPaid(OrderError):
    pass


@dataclass(frozen=True, slots=True)          # неизменяемость + экономия памяти
class Order:
    id: str
    customer_id: str
    total: Decimal                            # деньги — Decimal, никогда не float
    status: OrderStatus = OrderStatus.NEW

    def pay(self) -> Order:
        if self.status is OrderStatus.PAID:
            raise AlreadyPaid(self.id)        # инвариант живёт рядом с данными
        return replace(self, status=OrderStatus.PAID)

    def cancel(self) -> Order:
        if self.status is OrderStatus.PAID:
            raise CannotCancelPaid(self.id)
        return replace(self, status=OrderStatus.CANCELLED)

# src/orderflow/app/ports.py
from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from types import TracebackType
from typing import Protocol, runtime_checkable

from orderflow.domain.order import Order


class OrderRepository(Protocol):
    async def get(self, order_id: str) -> Order | None: ...
    async def save(self, order: Order) -> None: ...


class PaymentGateway(Protocol):
    async def charge(
        self, customer_id: str, amount: Decimal, idempotency_key: str
    ) -> str: ...


class Clock(Protocol):
    def now(self) -> datetime: ...     # время — тоже зависимость, иначе тесты флакают


@runtime_checkable                     # нужен, только если реально делаете isinstance
class EventPublisher(Protocol):
    async def publish(self, topic: str, payload: dict[str, object]) -> None: ...


class UnitOfWork(Protocol):
    """Границей транзакции управляет сценарий, а не репозиторий."""

    async def __aenter__(self) -> UnitOfWork: ...
    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> bool | None: ...
    async def commit(self) -> None: ...

# src/orderflow/app/use_cases/pay_order.py
from dataclasses import dataclass

from orderflow.app.ports import Clock, EventPublisher, OrderRepository, PaymentGateway, UnitOfWork
from orderflow.domain.order import Order


class OrderNotFound(Exception):
    pass


@dataclass(slots=True)
class PayOrder:
    orders: OrderRepository        # зависимости переданы, а не найдены импортом
    payments: PaymentGateway
    events: EventPublisher
    uow: UnitOfWork
    clock: Clock

    async def __call__(self, order_id: str, idempotency_key: str) -> Order:
        async with self.uow:                       # граница транзакции — здесь, не в репозитории
            order = await self.orders.get(order_id)
            if order is None:
                raise OrderNotFound(order_id)

            paid = order.pay()                     # доменное правило: может бросить AlreadyPaid
            await self.payments.charge(paid.customer_id, paid.total, idempotency_key)
            await self.orders.save(paid)
            await self.events.publish(             # outbox пишет в ту же транзакцию
                "order.paid",
                {"order_id": paid.id, "at": self.clock.now().isoformat()},
            )
            await self.uow.commit()
        return paid

# src/orderflow/adapters/postgres/orders.py
import asyncpg

from orderflow.domain.order import Order, OrderStatus


class PgOrderRepository:                      # НЕ наследует Protocol — совместимость структурная
    def __init__(self, conn: asyncpg.Connection) -> None:
        self._conn = conn

    async def get(self, order_id: str) -> Order | None:
        row = await self._conn.fetchrow(
            "SELECT id, customer_id, total, status FROM orders WHERE id = $1", order_id
        )
        if row is None:
            return None
        return Order(                          # маппинг строки в домен живёт в адаптере
            id=row["id"],
            customer_id=row["customer_id"],
            total=row["total"],
            status=OrderStatus(row["status"]),
        )

    async def save(self, order: Order) -> None:
        await self._conn.execute(
            "INSERT INTO orders (id, customer_id, total, status) VALUES ($1, $2, $3, $4)"
            " ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status",
            order.id, order.customer_id, order.total, order.status.value,
        )

# src/orderflow/adapters/postgres/orders.py — в конце файла
if TYPE_CHECKING:
    from orderflow.app.ports import OrderRepository

    _: type[OrderRepository] = PgOrderRepository   # mypy падёт здесь, если сигнатуры разошлись

# .importlinter
[importlinter]
root_packages = orderflow

[importlinter:contract:layers]
name = Слои: зависимости только вниз
type = layers
layers =
    orderflow.entrypoints
    orderflow.adapters
    orderflow.app
    orderflow.domain

[importlinter:contract:pure-domain]
name = Домен не знает про инфраструктуру
type = forbidden
source_modules =
    orderflow.domain
forbidden_modules =
    sqlalchemy
    httpx
    fastapi
    pydantic
    orderflow.adapters

[importlinter:contract:independent-entrypoints]
name = Входы не импортируют друг друга
type = independence
modules =
    orderflow.entrypoints.api
    orderflow.entrypoints.worker
    orderflow.entrypoints.cli

# tests/test_architecture.py
import ast
import pathlib

DOMAIN = pathlib.Path("src/orderflow/domain")
BANNED = {"sqlalchemy", "httpx", "fastapi", "orderflow.adapters", "orderflow.app"}


def _is_banned(module: str) -> bool:
    return any(module == b or module.startswith(f"{b}.") for b in BANNED)


def _imported_modules(path: pathlib.Path) -> set[str]:
    tree = ast.parse(path.read_text(encoding="utf-8"))
    names: set[str] = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            names.update(alias.name for alias in node.names)
        elif isinstance(node, ast.ImportFrom) and node.module and node.level == 0:
            names.add(node.module)
    return names


def test_domain_is_pure() -> None:
    for path in DOMAIN.rglob("*.py"):
        for module in _imported_modules(path):
            assert not _is_banned(module), f"{path}: запрещён импорт {module}"

from typing import TYPE_CHECKING

if TYPE_CHECKING:                       # во время выполнения импорта нет — цикл разорван
    from orderflow.app.services import Billing


def charge(billing: "Billing") -> None:  # либо from __future__ import annotations
    ...

# src/orderflow/bootstrap.py
from collections.abc import AsyncIterator, Callable
from contextlib import AsyncExitStack, asynccontextmanager
from dataclasses import dataclass

import asyncpg
import httpx

from orderflow.adapters.clock import SystemClock
from orderflow.adapters.events.outbox import OutboxPublisher
from orderflow.adapters.payments.http import HttpPaymentGateway
from orderflow.adapters.postgres.orders import PgOrderRepository
from orderflow.adapters.postgres.uow import PgUnitOfWork
from orderflow.app.use_cases.pay_order import PayOrder
from orderflow.config import Settings


@dataclass(slots=True, frozen=True)
class Container:
    """Плоский набор готовых сценариев. Никакой магии разрешения зависимостей."""
    pool: asyncpg.Pool
    pay_order_factory: Callable[[asyncpg.Connection], PayOrder]


@asynccontextmanager
async def build(settings: Settings) -> AsyncIterator[Container]:
    async with AsyncExitStack() as stack:     # закрытие в обратном порядке — гарантировано
        pool = await stack.enter_async_context(
            asyncpg.create_pool(
                dsn=str(settings.db.dsn),
                min_size=settings.db.pool_min,
                max_size=settings.db.pool_max,
                command_timeout=settings.db.command_timeout,
            )
        )
        http = await stack.enter_async_context(
            httpx.AsyncClient(
                base_url=str(settings.payments.base_url),
                timeout=httpx.Timeout(settings.payments.timeout),  # таймаут ВСЕГДА явный
            )
        )

        def pay_order_factory(conn: asyncpg.Connection) -> PayOrder:
            # соединение живёт один запрос, сценарий собирается вокруг него
            return PayOrder(
                orders=PgOrderRepository(conn),
                payments=HttpPaymentGateway(http, settings.payments.api_key),
                events=OutboxPublisher(conn),
                uow=PgUnitOfWork(conn),
                clock=SystemClock(),
            )

        yield Container(pool=pool, pay_order_factory=pay_order_factory)

# src/orderflow/entrypoints/api/routers/orders.py
from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import APIRouter, Depends, Header, HTTPException, Request, status

from orderflow.app.use_cases.pay_order import OrderNotFound, PayOrder
from orderflow.domain.order import AlreadyPaid
from orderflow.entrypoints.api.schemas import OrderResponse   # DTO транспорта, не домен

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


async def get_pay_order(request: Request) -> AsyncIterator[PayOrder]:
    container = request.app.state.container
    async with container.pool.acquire() as conn:      # одно соединение на запрос
        yield container.pay_order_factory(conn)


@router.post("/{order_id}/pay", status_code=status.HTTP_200_OK)
async def pay(
    order_id: str,
    idempotency_key: Annotated[str, Header(alias="Idempotency-Key")],
    use_case: Annotated[PayOrder, Depends(get_pay_order)],
) -> OrderResponse:
    try:
        order = await use_case(order_id, idempotency_key)
    except OrderNotFound:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "заказ не найден") from None
    except AlreadyPaid:
        raise HTTPException(status.HTTP_409_CONFLICT, "заказ уже оплачен") from None
    return OrderResponse.from_domain(order)           # доменный объект в DTO — здесь

# src/orderflow/config.py — единственный модуль, которому позволено знать про окружение
from typing import Literal

from pydantic import BaseModel, Field, HttpUrl, PostgresDsn, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class DatabaseSettings(BaseModel):
    dsn: PostgresDsn
    pool_min: int = Field(default=1, ge=1)
    pool_max: int = Field(default=10, ge=1, le=100)
    command_timeout: float = Field(default=5.0, gt=0)


class PaymentsSettings(BaseModel):
    base_url: HttpUrl
    api_key: SecretStr                       # не попадёт ни в repr, ни в лог, ни в трейсбек
    timeout: float = Field(default=3.0, gt=0)


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_prefix="APP_",
        env_nested_delimiter="__",           # APP_DB__POOL_MAX=20
        env_file=".env",                     # только dev: в образ файл не кладём
        secrets_dir="/run/secrets",          # docker/k8s secrets как файлы
        frozen=True,                         # конфиг не мутируют в рантайме
        extra="forbid",                      # опечатка APP_TIMEOUTT упадёт на старте
    )

    env: Literal["dev", "stage", "prod"] = "dev"
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
    log_format: Literal["json", "console"] = "json"
    db: DatabaseSettings
    payments: PaymentsSettings
    feature_new_pricing: bool = False        # безопасный дефолт — фича выключена

# ❌ ТАК НЕ НАДО: глобальный синглтон, читаемый на импорте
settings = Settings()                       # выполняется при import config

# в любом модуле:
from orderflow.config import settings       # неявная зависимость всего от всего

# ⚠️ КОМПРОМИСС: работает, но требует дисциплины
from functools import lru_cache

@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()                       # в тестах не забыть get_settings.cache_clear()

# src/orderflow/logging_setup.py
import contextvars
import json
import logging
import logging.config
from datetime import UTC, datetime

request_id: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")

# набор «служебных» полей LogRecord — всё остальное считаем пользовательским
_RESERVED = frozenset(
    logging.LogRecord("", 0, "", 0, "", (), None).__dict__
) | {"message", "asctime", "taskName"}


class ContextFilter(logging.Filter):
    """Проставляет сквозные поля из contextvars в каждую запись."""

    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id.get()
        return True


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, object] = {
            "ts": datetime.fromtimestamp(record.created, UTC).isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "logger": record.name,
            "msg": record.getMessage(),
            "request_id": getattr(record, "request_id", "-"),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        # всё, что пришло через extra=..., едет отдельными полями
        payload.update({k: v for k, v in record.__dict__.items() if k not in _RESERVED})
        return json.dumps(payload, ensure_ascii=False, default=str)


def configure(level: str, fmt: str) -> None:
    logging.config.dictConfig({
        "version": 1,
        "disable_existing_loggers": False,          # иначе логгеры библиотек замолчат
        "filters": {"context": {"()": ContextFilter}},
        "formatters": {
            "json": {"()": JsonFormatter},
            "console": {"format": "%(asctime)s %(levelname)-7s %(name)s [%(request_id)s] %(message)s"},
        },
        "handlers": {
            "stdout": {
                "class": "logging.StreamHandler",
                "stream": "ext://sys.stdout",
                "formatter": fmt,
                "filters": ["context"],
            }
        },
        "root": {"level": level, "handlers": ["stdout"]},
        "loggers": {                                 # чужой шум приглушаем точечно
            "uvicorn.access": {"level": "WARNING"},
            "httpx": {"level": "WARNING"},
            "asyncio": {"level": "WARNING"},
        },
    })
    logging.captureWarnings(True)                    # DeprecationWarning тоже попадут в логи

logger = logging.getLogger(__name__)

logger.info(
    "order_paid",                                   # СОБЫТИЕ, а не предложение на русском
    extra={"order_id": order.id, "amount": str(order.total), "duration_ms": 37},
)
# {"ts":"2026-07-16T10:00:00.123+00:00","level":"INFO","logger":"orderflow.app.use_cases",
#  "msg":"order_paid","request_id":"7f3c…","order_id":"42","amount":"199.00","duration_ms":37}

# src/orderflow/entrypoints/api/main.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from orderflow import bootstrap, logging_setup
from orderflow.config import Settings
from orderflow.domain.order import OrderError
from orderflow.entrypoints.api.errors import domain_error_handler
from orderflow.entrypoints.api.routers import orders


def create_app(settings: Settings | None = None) -> FastAPI:
    settings = settings or Settings()             # тест может подсунуть свой конфиг
    logging_setup.configure(settings.log_level, settings.log_format)

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        async with bootstrap.build(settings) as container:
            app.state.container = container       # ресурсы живут ровно столько, сколько приложение
            yield
        # выход из async with = корректное закрытие пулов при остановке

    app = FastAPI(title="orderflow", lifespan=lifespan)
    app.include_router(orders.router)
    app.add_exception_handler(OrderError, domain_error_handler)
    return app

import asyncio
import logging
import sys
import threading

log = logging.getLogger("unhandled")

sys.excepthook = lambda *a: log.critical("unhandled_exception", exc_info=a)
threading.excepthook = lambda args: log.critical("thread_exception", exc_info=args[:3])
asyncio.get_running_loop().set_exception_handler(
    lambda loop, ctx: log.error("asyncio_exception", extra={"detail": str(ctx.get("message"))})
)

# tests/fakes.py — дубль вместо мока: mypy проверит его совместимость с портом
class InMemoryOrderRepository:
    def __init__(self) -> None:
        self._items: dict[str, Order] = {}

    async def get(self, order_id: str) -> Order | None:
        return self._items.get(order_id)

    async def save(self, order: Order) -> None:
        self._items[order.id] = order


# tests/test_pay_order.py
async def test_pay_order_marks_paid() -> None:
    repo = InMemoryOrderRepository()
    await repo.save(Order(id="42", customer_id="c1", total=Decimal("199.00")))
    use_case = PayOrder(orders=repo, payments=FakeGateway(), events=NullPublisher(),
                        uow=NullUow(), clock=FrozenClock())

    result = await use_case("42", idempotency_key="k1")

    assert result.status is OrderStatus.PAID          # ни базы, ни сети, ни моков