Применяешь ли чистую архитектуру
Комментарии (1)
Ответ сгенерирован нейросетью и может содержать ошибки
# Применение чистой архитектуры в проектах
Введение
Чистая архитектура (Clean Architecture) — это принцип организации кода, предложенный Робертом Мартином (Uncle Bob), который нацелен на создание систем, независимых от фреймворков, тестируемых, масштабируемых и простых в поддержке. Я активно применяю эти принципы в своих проектах.
Слои чистой архитектуры
Структура слоёв (от внешних к внутренним):
┌─────────────────────────────────────────┐
│ Presentation (UI, API, CLI) │
├─────────────────────────────────────────┤
│ Application (Use Cases, DTOs) │
├─────────────────────────────────────────┤
│ Domain (Entities, Value Objects) │
├─────────────────────────────────────────┤
│ Infrastructure (DB, APIs, Services) │
└─────────────────────────────────────────┘
Правило: зависимости только внутрь →
Domain Layer (Сердце системы)
Это слой бизнес-логики, полностью независимый от деталей реализации:
# domain/entities/user.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@dataclass
class User:
"""User Entity - чистая бизнес-логика"""
id: str
email: str
password_hash: str
is_active: bool
created_at: datetime
def is_valid_email(self) -> bool:
"""Проверка корректности email"""
return "@" in self.email and "." in self.email.split("@")[1]
def is_password_strong(self, password: str) -> bool:
"""Проверка надёжности пароля"""
return len(password) >= 8 and any(c.isupper() for c in password)
def can_login(self) -> bool:
"""Может ли пользователь войти"""
return self.is_active
# domain/value_objects/email.py
from dataclasses import dataclass
@dataclass(frozen=True)
class Email:
"""Value Object для email - неизменяемый"""
value: str
def __post_init__(self):
if "@" not in self.value:
raise ValueError("Некорректный email")
def __str__(self) -> str:
return self.value
# domain/repositories/user_repository.py
from abc import ABC, abstractmethod
from typing import Optional
from .entities import User
class UserRepository(ABC):
"""Абстракция для работы с пользователями"""
@abstractmethod
async def get_by_id(self, user_id: str) -> Optional[User]:
"""Получить пользователя по ID"""
pass
@abstractmethod
async def save(self, user: User) -> None:
"""Сохранить пользователя"""
pass
@abstractmethod
async def delete(self, user_id: str) -> None:
"""Удалить пользователя"""
pass
# domain/services/password_service.py
import hashlib
import secrets
class PasswordService:
"""Сервис для работы с паролями"""
@staticmethod
def hash_password(password: str) -> str:
"""Хешировать пароль"""
salt = secrets.token_hex(16)
hashed = hashlib.pbkdf2_hmac(
sha256,
password.encode(),
salt.encode(),
100000
)
return f"{salt}${hashed.hex()}"
@staticmethod
def verify_password(password: str, hash_value: str) -> bool:
"""Проверить пароль"""
salt, _ = hash_value.split("$")
hashed = hashlib.pbkdf2_hmac(
sha256,
password.encode(),
salt.encode(),
100000
)
return hash_value == f"{salt}${hashed.hex()}"
Application Layer (Бизнес-процессы)
Этот слой содержит use cases - сценарии использования приложения:
# application/dtos/user_dto.py
from dataclasses import dataclass
@dataclass
class CreateUserDTO:
"""DTO для создания пользователя"""
email: str
password: str
@dataclass
class UserResponseDTO:
"""DTO для ответа"""
id: str
email: str
is_active: bool
# application/use_cases/create_user.py
from typing import Optional
from domain.entities import User
from domain.repositories import UserRepository
from domain.services import PasswordService
from domain.value_objects import Email
from .dtos import CreateUserDTO, UserResponseDTO
class CreateUserUseCase:
"""Use case для создания пользователя"""
def __init__(self, user_repo: UserRepository):
self.user_repo = user_repo
self.password_service = PasswordService()
async def execute(self, dto: CreateUserDTO) -> UserResponseDTO:
"""Выполнить use case"""
# Валидация
try:
email = Email(dto.email)
except ValueError as e:
raise ValueError(f"Некорректный email: {e}")
# Проверка существования
existing_user = await self.user_repo.get_by_email(dto.email)
if existing_user:
raise ValueError("Пользователь с таким email уже существует")
# Создание сущности
user = User(
id=self._generate_id(),
email=dto.email,
password_hash=self.password_service.hash_password(dto.password),
is_active=True,
created_at=datetime.now()
)
# Сохранение
await self.user_repo.save(user)
# Возврат DTO
return UserResponseDTO(
id=user.id,
email=user.email,
is_active=user.is_active
)
@staticmethod
def _generate_id() -> str:
import uuid
return str(uuid.uuid4())
# application/use_cases/authenticate_user.py
class AuthenticateUserUseCase:
"""Use case для аутентификации"""
def __init__(self, user_repo: UserRepository):
self.user_repo = user_repo
self.password_service = PasswordService()
async def execute(self, email: str, password: str) -> Optional[UserResponseDTO]:
user = await self.user_repo.get_by_email(email)
if not user:
return None
if not self.password_service.verify_password(password, user.password_hash):
return None
if not user.can_login():
return None
return UserResponseDTO(
id=user.id,
email=user.email,
is_active=user.is_active
)
Infrastructure Layer (Реализация деталей)
Здесь реализуются абстракции из domain слоя:
# infrastructure/repositories/postgres_user_repository.py
from typing import Optional
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from domain.entities import User
from domain.repositories import UserRepository
from .models import UserModel
class PostgresUserRepository(UserRepository):
"""Реализация UserRepository для PostgreSQL"""
def __init__(self, session: AsyncSession):
self.session = session
async def get_by_id(self, user_id: str) -> Optional[User]:
stmt = select(UserModel).where(UserModel.id == user_id)
result = await self.session.execute(stmt)
model = result.scalar_one_or_none()
return self._to_entity(model) if model else None
async def get_by_email(self, email: str) -> Optional[User]:
stmt = select(UserModel).where(UserModel.email == email)
result = await self.session.execute(stmt)
model = result.scalar_one_or_none()
return self._to_entity(model) if model else None
async def save(self, user: User) -> None:
model = self._to_model(user)
self.session.add(model)
await self.session.flush()
@staticmethod
def _to_entity(model: UserModel) -> User:
return User(
id=model.id,
email=model.email,
password_hash=model.password_hash,
is_active=model.is_active,
created_at=model.created_at
)
@staticmethod
def _to_model(user: User) -> UserModel:
return UserModel(
id=user.id,
email=user.email,
password_hash=user.password_hash,
is_active=user.is_active,
created_at=user.created_at
)
# infrastructure/models.py
from sqlalchemy import Column, String, Boolean, DateTime
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class UserModel(Base):
__tablename__ = "users"
id = Column(String, primary_key=True)
email = Column(String, unique=True, nullable=False)
password_hash = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, nullable=False)
Presentation Layer (API/UI)
Зависит от всех слоёв, но не наоборот:
# presentation/api/routers/users.py
from fastapi import APIRouter, Depends, HTTPException
from application.dtos import CreateUserDTO, UserResponseDTO
from application.use_cases import CreateUserUseCase, AuthenticateUserUseCase
from presentation.dependencies import get_user_repository
router = APIRouter(prefix="/users", tags=["users"])
@router.post("/register", response_model=UserResponseDTO)
async def register(dto: CreateUserDTO, repo = Depends(get_user_repository)):
use_case = CreateUserUseCase(repo)
try:
result = await use_case.execute(dto)
return result
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
@router.post("/login", response_model=UserResponseDTO)
async def login(email: str, password: str, repo = Depends(get_user_repository)):
use_case = AuthenticateUserUseCase(repo)
result = await use_case.execute(email, password)
if not result:
raise HTTPException(status_code=401, detail="Invalid credentials")
return result
Структура проекта
project/
├── domain/
│ ├── entities/
│ │ └── user.py
│ ├── value_objects/
│ │ └── email.py
│ ├── repositories/
│ │ └── user_repository.py
│ └── services/
│ └── password_service.py
├── application/
│ ├── dtos/
│ │ └── user_dto.py
│ └── use_cases/
│ ├── create_user.py
│ └── authenticate_user.py
├── infrastructure/
│ ├── repositories/
│ │ └── postgres_user_repository.py
│ ├── models.py
│ └── services/
│ └── email_service.py
├── presentation/
│ ├── api/
│ │ ├── routers/
│ │ │ └── users.py
│ │ └── main.py
│ └── dependencies.py
└── tests/
├── unit/
├── integration/
└── e2e/
Преимущества, которые я получаю
✅ Независимость от фреймворков — код не привязан к FastAPI, SQLAlchemy и т.д. ✅ Тестируемость — use cases тестируются с mock-репозиториями ✅ Масштабируемость — легко добавлять новые use cases ✅ Переиспользуемость — одинаковые use cases для разных фронтов ✅ Поддерживаемость — ясная организация, лёгкое изменение деталей
Практический пример тестирования
# tests/unit/use_cases/test_create_user.py
import pytest
from unittest.mock import AsyncMock
from application.use_cases import CreateUserUseCase
from application.dtos import CreateUserDTO
from domain.repositories import UserRepository
@pytest.mark.asyncio
async def test_create_user_success():
# Arrange
mock_repo = AsyncMock(spec=UserRepository)
mock_repo.get_by_email.return_value = None
use_case = CreateUserUseCase(mock_repo)
dto = CreateUserDTO(email="test@example.com", password="SecurePass123")
# Act
result = await use_case.execute(dto)
# Assert
assert result.email == "test@example.com"
mock_repo.save.assert_called_once()
@pytest.mark.asyncio
async def test_create_user_duplicate_email():
# Arrange
mock_repo = AsyncMock(spec=UserRepository)
mock_repo.get_by_email.return_value = User(...) # Пользователь существует
use_case = CreateUserUseCase(mock_repo)
dto = CreateUserDTO(email="test@example.com", password="SecurePass123")
# Act & Assert
with pytest.raises(ValueError):
await use_case.execute(dto)
Выводы
Чистая архитектура — это не просто теория, а практический подход, который я применяю ежедневно:
- Разделение слоёв упрощает разработку и тестирование
- Абстракции в domain слое позволяют легко менять реализацию
- Use cases четко описывают бизнес-процессы
- Код становится более гибким, тестируемым и масштабируемым
- Новые разработчики быстрее ориентируются в проекте благодаря чёткой организации