Какие знаешь архитектурные шаблоны?

Архитектурные шаблоны (Architectural Patterns)

Архитектурные шаблоны определяют структуру приложения и способ организации компонентов. Разберём самые важные из них для Python разработчика.

1. Многоуровневая архитектура (Layered Architecture)

Классическая структура с разделением на слои. Каждый слой отвечает за определённый аспект.

# Структура проекта
app/
├── presentation/          # Слой представления (контроллеры, API)
│   ├── routes.py
│   └── schemas.py
├── application/           # Слой приложения (use cases, бизнес-логика)
│   ├── user_service.py
│   └── dto/
├── domain/               # Слой домена (сущности, интерфейсы)
│   ├── user.py
│   └── repositories.py
└── infrastructure/       # Слой инфраструктуры (БД, кеш, API)
    ├── database.py
    └── cache.py

# Зависимости идут только вниз (presentation → application → domain)
# presentation НЕ может использовать infrastructure напрямую

Пример кода:

# domain/user.py
class User:
    def __init__(self, id: int, name: str, email: str):
        self.id = id
        self.name = name
        self.email = email

class UserRepository:
    def get_by_id(self, user_id: int) -> User:
        raise NotImplementedError

# application/user_service.py
class UserService:
    def __init__(self, repository: UserRepository):
        self.repository = repository
    
    def get_user(self, user_id: int) -> User:
        return self.repository.get_by_id(user_id)

# infrastructure/database.py
class DatabaseUserRepository(UserRepository):
    def __init__(self, db_connection):
        self.db = db_connection
    
    def get_by_id(self, user_id: int) -> User:
        result = self.db.query("SELECT * FROM users WHERE id = ?", (user_id,))
        return User(result['id'], result['name'], result['email'])

# presentation/routes.py
from fastapi import FastAPI

app = FastAPI()
repository = DatabaseUserRepository(db_connection)
service = UserService(repository)

@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = service.get_user(user_id)
    return {"id": user.id, "name": user.name, "email": user.email}

Преимущества:

• Простота понимания
• Хорошая разделение ответственности
• Легко тестировать каждый слой

Недостатки:

• Может стать громоздкой на больших проектах
• Зависимость между слоями

2. Микросервисная архитектура (Microservices)

Приложение разделено на независимые сервисы, каждый решает одну задачу.

# Структура
services/
├── user_service/
│   ├── app.py
│   ├── models.py
│   └── routes.py
├── order_service/
│   ├── app.py
│   ├── models.py
│   └── routes.py
└── payment_service/
    ├── app.py
    ├── models.py
    └── routes.py

# Каждый сервис:
# - Имеет свою БД
# - Использует REST/gRPC/Message Queue для общения
# - Может быть развёрнут независимо

Пример взаимодействия:

# user_service/app.py
from fastapi import FastAPI
import httpx

app = FastAPI()

@app.post("/users")
async def create_user(user_data: dict):
    # Создаём пользователя
    user = db.insert(user_data)
    
    # Отправляем событие в другой сервис
    async with httpx.AsyncClient() as client:
        await client.post(
            "http://order_service/events/user_created",
            json={"user_id": user.id, "email": user.email}
        )
    
    return user

# order_service/app.py
@app.post("/events/user_created")
async def on_user_created(event: dict):
    # Когда создаётся пользователь, готовим приветственное предложение
    send_welcome_offer(event['user_id'], event['email'])
    return {"status": "processed"}

Преимущества:

• Независимое масштабирование
• Разные технологии для разных сервисов
• Отказоустойчивость (один сервис упал — остальные работают)

Недостатки:

• Сложная отладка
• Сетевые задержки
• Консистентность данных (eventual consistency)

3. Event-Driven архитектура

Компоненты общаются через события, а не прямыми вызовами.

# event_bus.py
from typing import Callable, List

class EventBus:
    def __init__(self):
        self.handlers: dict[str, List[Callable]] = {}
    
    def subscribe(self, event_type: str, handler: Callable):
        if event_type not in self.handlers:
            self.handlers[event_type] = []
        self.handlers[event_type].append(handler)
    
    def publish(self, event_type: str, data: dict):
        if event_type in self.handlers:
            for handler in self.handlers[event_type]:
                handler(data)

event_bus = EventBus()

# domain/events.py
class UserCreatedEvent:
    def __init__(self, user_id: int, email: str):
        self.user_id = user_id
        self.email = email

# application/user_service.py
class UserService:
    def __init__(self, repository, event_bus: EventBus):
        self.repository = repository
        self.event_bus = event_bus
    
    def create_user(self, name: str, email: str) -> int:
        user = self.repository.create(name, email)
        # Публикуем событие вместо прямого вызова
        self.event_bus.publish('user.created', {
            'user_id': user.id,
            'email': user.email
        })
        return user.id

# Другие сервисы слушают события
def send_welcome_email(event_data):
    print(f"Отправляем письмо на {event_data['email']}")

event_bus.subscribe('user.created', send_welcome_email)

4. Clean Architecture (Чистая архитектура)

Эволюция многоуровневой архитектуры с чётким разделением ответственности.

# Структура
app/
├── entities/              # Бизнес-правила (ядро)
│   └── user.py
├── use_cases/             # Сценарии использования
│   └── create_user.py
├── interface_adapters/    # Адаптеры (преобразование данных)
│   ├── repositories.py
│   ├── presenters.py
│   └── gateways.py
└── frameworks/            # Фреймворки и инструменты
    ├── web/
    ├── db/
    └── cache/

# Правило: зависимости направлены к центру (entities)

Пример:

# entities/user.py
class User:
    def __init__(self, name: str, email: str):
        self.name = name
        self.email = email
    
    def is_valid(self) -> bool:
        return "@" in self.email and len(self.name) > 0

# use_cases/create_user.py
class CreateUserUseCase:
    def __init__(self, repository):
        self.repository = repository
    
    def execute(self, name: str, email: str) -> int:
        user = User(name, email)
        if not user.is_valid():
            raise ValueError("Invalid user data")
        return self.repository.save(user)

# interface_adapters/repositories.py
class UserRepository:
    def save(self, user: User) -> int:
        result = db.insert({
            'name': user.name,
            'email': user.email
        })
        return result['id']

# frameworks/web/routes.py
@app.post("/users")
async def create_user(data: dict):
    use_case = CreateUserUseCase(UserRepository())
    user_id = use_case.execute(data['name'], data['email'])
    return {"user_id": user_id}

5. CQRS (Command Query Responsibility Segregation)

Разделение операций на команды (изменение) и запросы (чтение).

# Модели
class CreateUserCommand:
    def __init__(self, name: str, email: str):
        self.name = name
        self.email = email

class GetUserQuery:
    def __init__(self, user_id: int):
        self.user_id = user_id

# Обработчики
class CreateUserCommandHandler:
    def handle(self, command: CreateUserCommand) -> int:
        user = User(command.name, command.email)
        db.save(user)
        # Синхронизируем с read моделью
        read_db.insert({'id': user.id, 'name': user.name, 'email': user.email})
        return user.id

class GetUserQueryHandler:
    def handle(self, query: GetUserQuery) -> dict:
        # Читаем из оптимизированной read модели
        return read_db.get_by_id(query.user_id)

# API
@app.post("/users")
async def create_user(data: dict):
    command = CreateUserCommand(data['name'], data['email'])
    user_id = CreateUserCommandHandler().handle(command)
    return {"user_id": user_id}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    query = GetUserQuery(user_id)
    user = GetUserQueryHandler().handle(query)
    return user

6. Hexagonal Architecture (Портов и адаптеров)

Приложение окружено портами (интерфейсами) и адаптерами для взаимодействия с внешним миром.

# core/user_service.py (ядро приложения)
class UserService:
    def __init__(self, repository, notification_gateway):
        self.repository = repository
        self.notification_gateway = notification_gateway
    
    def create_user(self, name: str, email: str):
        user = User(name, email)
        self.repository.save(user)
        self.notification_gateway.send_welcome_email(email)
        return user

# ports/repository.py (портов)
class RepositoryPort:
    def save(self, user):
        raise NotImplementedError

class NotificationPort:
    def send_welcome_email(self, email: str):
        raise NotImplementedError

# adapters/database.py (адаптеры)
class PostgresRepository(RepositoryPort):
    def save(self, user):
        db.insert(user)

class EmailNotificationAdapter(NotificationPort):
    def send_welcome_email(self, email: str):
        send_email(email, "Welcome!")

Сравнение архитектур

Выводы

Основные архитектурные шаблоны:

1. Layered — для простых и средних проектов
2. Microservices — для больших распределённых систем
3. Event-Driven — для асинхронных и слабо связанных систем
4. Clean Architecture — для проектов любого размера с хорошей архитектурой
5. CQRS — для систем с разными требованиями к чтению и записи
6. Hexagonal — для максимальной гибкости и тестируемости

Выбор архитектуры зависит от размера проекта, команды, требований к производительности и плана развития.