Как ведет себя поток управления в чистой архитектуре?

Как ведет себя поток управления в чистой архитектуре

Поток управления (Control Flow) в Clean Architecture — это основной механизм, который определяет, как данные и управление движутся между слоями. Это критично для понимания архитектуры.

Принцип Dependency Inversion

В Clean Architecture есть одно правило: зависимости всегда указывают внутрь (к бизнес-логике), никогда наружу.

Presentation (UI) 
      ↓ зависит
Application (Use Cases)
      ↓ зависит
Domain (Entities, Business Logic)
      ← Domain никогда не зависит ни от чего!

Слои архитектуры

1. Domain Layer (самый внутренний)

Это ядро приложения — чистая бизнес-логика, независимая от фреймворков и UI.

# src/domain/entities.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    """Сущность User — чистая бизнес-логика"""
    id: str
    email: str
    password_hash: str
    created_at: datetime
    
    def is_password_valid(self, password: str, hasher) -> bool:
        """Бизнес-логика проверки пароля"""
        return hasher.verify(password, self.password_hash)

@dataclass
class AuthResult:
    """Результат аутентификации"""
    success: bool
    user: User | None
    error: str | None

# src/domain/interfaces.py
from abc import ABC, abstractmethod

class UserRepository(ABC):
    """Интерфейс репозитория — Domain определяет контракт!"""
    @abstractmethod
    async def find_by_email(self, email: str) -> User | None:
        pass
    
    @abstractmethod
    async def save(self, user: User) -> None:
        pass

class PasswordHasher(ABC):
    """Domain зависит от интерфейса, не от реализации"""
    @abstractmethod
    def hash(self, password: str) -> str:
        pass
    
    @abstractmethod
    def verify(self, password: str, hash_: str) -> bool:
        pass

Ключевой момент: Domain не импортирует ничего, кроме встроенных типов Python. Ноль зависимостей!

2. Application Layer (Use Cases)

Это оркестраторы — они используют Domain сущности и интерфейсы для реализации бизнес-процессов.

# src/application/auth/login_use_case.py
from src.domain.entities import AuthResult, User
from src.domain.interfaces import UserRepository, PasswordHasher

class LoginUseCase:
    """Use Case — это бизнес-процесс"""
    
    def __init__(
        self,
        user_repository: UserRepository,
        password_hasher: PasswordHasher,
        jwt_service: "JWTService"
    ):
        """Dependency Injection — зависимости приходят в конструктор"""
        self.user_repo = user_repository
        self.hasher = password_hasher
        self.jwt_service = jwt_service
    
    async def execute(self, email: str, password: str) -> AuthResult:
        """Главная логика Use Case"""
        # Шаг 1: Найти пользователя
        user = await self.user_repo.find_by_email(email)
        
        if not user:
            return AuthResult(
                success=False,
                user=None,
                error="User not found"
            )
        
        # Шаг 2: Проверить пароль (Domain логика)
        if not user.is_password_valid(password, self.hasher):
            return AuthResult(
                success=False,
                user=None,
                error="Invalid password"
            )
        
        # Шаг 3: Создать токен
        token = self.jwt_service.generate_token(user.id)
        
        return AuthResult(
            success=True,
            user=user,
            error=None
        )

Ключевой момент: Use Case знает про Domain и интерфейсы, но ничего не знает о FastAPI, PostgreSQL, Redis — всё это приходит через DI.

3. Infrastructure Layer

Здесь реализуются интерфейсы Domain. PostgreSQL, Redis, third-party APIs.

# src/infrastructure/persistence/user_repository.py
from sqlalchemy.orm import Session
from src.domain.entities import User
from src.domain.interfaces import UserRepository
from src.infrastructure.db.models import UserModel

class PostgresUserRepository(UserRepository):
    """Реализация интерфейса для PostgreSQL"""
    
    def __init__(self, db: Session):
        self.db = db
    
    async def find_by_email(self, email: str) -> User | None:
        # SQL запрос
        user_model = self.db.query(UserModel).filter(
            UserModel.email == email
        ).first()
        
        if not user_model:
            return None
        
        # Преобразуем Model → Domain Entity
        return User(
            id=user_model.id,
            email=user_model.email,
            password_hash=user_model.password_hash,
            created_at=user_model.created_at
        )
    
    async def save(self, user: User) -> None:
        user_model = UserModel(
            id=user.id,
            email=user.email,
            password_hash=user.password_hash,
            created_at=user.created_at
        )
        self.db.add(user_model)
        self.db.commit()

# src/infrastructure/security/bcrypt_hasher.py
from bcrypt import hashpw, checkpw
from src.domain.interfaces import PasswordHasher

class BcryptPasswordHasher(PasswordHasher):
    """Реализация хеширования паролей"""
    
    def hash(self, password: str) -> str:
        return hashpw(password.encode(), bcrypt.gensalt()).decode()
    
    def verify(self, password: str, hash_: str) -> bool:
        return checkpw(password.encode(), hash_.encode())

Ключевой момент: Infrastructure реализует Domain интерфейсы. Если завтра нужна MongoDB вместо PostgreSQL, меняешь только эту папку.

4. Presentation Layer (REST API, CLI, WebSocket)

Задача: получить HTTP запрос → вызвать Use Case → вернуть JSON.

# src/presentation/api/auth_routes.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from src.application.auth.login_use_case import LoginUseCase
from src.infrastructure.di.container import get_login_use_case

router = APIRouter(prefix="/api/v1/auth", tags=["auth"])

# Request/Response модели (Presentation слой)
class LoginRequest(BaseModel):
    email: str
    password: str

class LoginResponse(BaseModel):
    access_token: str
    token_type: str = "Bearer"
    user_id: str

class ErrorResponse(BaseModel):
    error: str

@router.post("/login")
async def login(
    request: LoginRequest,
    use_case: LoginUseCase = Depends(get_login_use_case)
) -> LoginResponse:
    """Endpoint — это про конвертацию HTTP ↔ Domain"""
    
    # Вызываем Use Case
    result = await use_case.execute(request.email, request.password)
    
    # Если успех — возвращаем ответ
    if result.success:
        return LoginResponse(
            access_token=result.token,
            user_id=result.user.id
        )
    
    # Если ошибка — возвращаем 401
    raise HTTPException(status_code=401, detail=result.error)

Ключевой момент: Presentation ничего не знает про базу данных, хеширование, бизнес-логику. Всё делегирует Use Case.

Поток управления: детально

HTTP запрос: POST /api/v1/auth/login
                        ↓
            Presentation Layer
     (LoginRequest → валидация Pydantic)
                        ↓
         Получить Use Case через DI
                        ↓
      Application Layer (LoginUseCase)
    execute(email, password) → вызывает Domain логику
                        ↓
              Domain Layer
         User.is_password_valid()
              (чистая бизнес-логика)
                        ↓
        Запрос к Infrastructure
     (UserRepository.find_by_email)
                        ↓
        Infrastructure Layer
      (PostgresUserRepository)
          SELECT * FROM users WHERE email=?
                        ↓
             Database (PostgreSQL)
                        ↓
        Infrastructure возвращает User
                        ↓
       Domain выполняет is_password_valid
                        ↓
     Use Case генерирует token, возвращает AuthResult
                        ↓
       Presentation конвертирует в LoginResponse
                        ↓
        HTTP ответ: {"access_token": "...", ...}

Правило: Dependency Inversion в действии

# ❌ НЕПРАВИЛЬНО — Presentation зависит от Infrastructure
from src.infrastructure.db import get_db_session  # Зависит от DB!
from sqlalchemy.orm import Session

def login(request: LoginRequest, db: Session):
    # Логика здесь — нарушает архитектуру!
    user = db.query(User).filter(User.email == request.email).first()

# ✅ ПРАВИЛЬНО — всё через интерфейсы и DI
from src.application.auth.login_use_case import LoginUseCase

def login(request: LoginRequest, use_case: LoginUseCase = Depends()):
    result = await use_case.execute(request.email, request.password)
    # Presentation ничего не знает про БД!

Dependency Injection Container

Всё это работает благодаря DI контейнеру:

# src/infrastructure/di/container.py
from src.application.auth.login_use_case import LoginUseCase
from src.infrastructure.persistence.user_repository import PostgresUserRepository
from src.infrastructure.security.bcrypt_hasher import BcryptPasswordHasher

def get_login_use_case(db: Session = Depends(get_db)) -> LoginUseCase:
    """Собираем Use Case с инъекциями"""
    user_repo = PostgresUserRepository(db)
    hasher = BcryptPasswordHasher()
    jwt_service = JWTService()
    
    return LoginUseCase(user_repo, hasher, jwt_service)

Преимущество: Если завтра меняешь реализацию UserRepository с Postgres на MongoDB — меняешь только одну строку в контейнере!

Заключение

Поток управления в Clean Architecture:

1. Request входит в Presentation
2. Presentation вызывает Use Case через DI
3. Use Case использует Domain интерфейсы
4. Infrastructure реализует эти интерфейсы
5. Response выходит из Presentation

Ключо: Domain никогда не знает про детали реализации, только про интерфейсы. Это делает архитектуру гибкой и тестируемой.