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

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

Архитектурный стиль определяет общую структуру приложения, способ организации компонентов и их взаимодействия. Знание различных стилей критично для проектирования масштабируемых и поддерживаемых систем.

1. Монолитная архитектура (Monolithic)

Всё приложение - это один большой компонент, разделённый на модули.

# Структура монолита:
# app/
# ├── models/
# │   ├── user.py
# │   └── product.py
# ├── services/
# │   ├── user_service.py
# │   └── product_service.py
# ├── routes/
# │   ├── user_routes.py
# │   └── product_routes.py
# └── main.py

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

@app.get("/products/{product_id}")
def get_product(product_id: int):
    return {"product_id": product_id}

Плюсы: простота разработки, легко отлаживать, хорошая производительность.

Минусы: масштабирование, развёртывание, технологический долг.

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

Приложение разделено на несколько независимых сервисов, каждый отвечает за одну функцию.

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

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id, "name": "Alice"}

# product_service/main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/products/{product_id}")
async def get_product(product_id: int):
    return {"product_id": product_id, "name": "Laptop"}

# api_gateway/main.py
from fastapi import FastAPI
import httpx

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    async with httpx.AsyncClient() as client:
        user = await client.get(f"http://user-service/users/{user_id}")
    return user.json()

@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    async with httpx.AsyncClient() as client:
        order = await client.get(f"http://order-service/orders/{order_id}")
    return order.json()

Плюсы: независимое масштабирование, независимые развёртывания, разные технологии.

Минусы: сложность распределённой системы, сетевые задержки, консистентность.

3. Слойная архитектура (Layered/N-Tier)

Приложение разделено на горизонтальные слои: presentation, business logic, persistence.

# presentation слой
from fastapi import FastAPI
from app.services.user_service import UserService

app = FastAPI()
user_service = UserService()

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

# business logic слой (services/user_service.py)
from app.repositories.user_repository import UserRepository

class UserService:
    def __init__(self):
        self.repository = UserRepository()
    
    def get_user(self, user_id: int):
        user = self.repository.find_by_id(user_id)
        if not user:
            raise ValueError("User not found")
        return user
    
    def create_user(self, name: str, email: str):
        if not self._validate_email(email):
            raise ValueError("Invalid email")
        user = User(name=name, email=email)
        return self.repository.save(user)
    
    def _validate_email(self, email: str) -> bool:
        return "@" in email

# persistence слой (repositories/user_repository.py)
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

class UserRepository:
    def __init__(self):
        self.engine = create_engine("postgresql://...")
    
    def find_by_id(self, user_id: int):
        with Session(self.engine) as session:
            return session.query(User).filter(User.id == user_id).first()
    
    def save(self, user):
        with Session(self.engine) as session:
            session.add(user)
            session.commit()
        return user

Плюсы: разделение ответственности, переиспользуемость, тестируемость.

Минусы: может быть излишняя абстракция, производительность.

4. Clean Architecture

Центр архитектуры - domain logic, зависимости указывают внутрь (presentation → application → domain).

# domain/user/user.py - сущности
class User:
    def __init__(self, id: int, name: str, email: str):
        self.id = id
        self.name = name
        self.email = email
    
    def update_email(self, new_email: str):
        if "@" not in new_email:
            raise ValueError("Invalid email")
        self.email = new_email

# application/user/user_service.py - use cases
from domain.user.user import User
from application.user.user_repository import UserRepository

class UpdateUserEmailUseCase:
    def __init__(self, repository: UserRepository):
        self.repository = repository
    
    def execute(self, user_id: int, new_email: str) -> User:
        user = self.repository.find_by_id(user_id)
        if not user:
            raise ValueError("User not found")
        user.update_email(new_email)
        self.repository.save(user)
        return user

# infrastructure/persistence/user_repository_impl.py
from sqlalchemy.orm import Session
from application.user.user_repository import UserRepository

class SQLAlchemyUserRepository(UserRepository):
    def __init__(self, session: Session):
        self.session = session
    
    def find_by_id(self, user_id: int):
        return self.session.query(UserModel).filter(UserModel.id == user_id).first()
    
    def save(self, user):
        self.session.add(user)
        self.session.commit()

# presentation/routes/user_routes.py
from fastapi import APIRouter
from application.user.update_user_email_use_case import UpdateUserEmailUseCase

router = APIRouter()

@router.put("/users/{user_id}/email")
def update_user_email(user_id: int, new_email: str, use_case: UpdateUserEmailUseCase):
    updated_user = use_case.execute(user_id, new_email)
    return {"id": updated_user.id, "email": updated_user.email}

Плюсы: независимость от фреймворков, тестируемость, поддерживаемость.

Минусы: сложность, кривая обучения.

5. Событийно-ориентированная архитектура (Event-Driven)

Компоненты взаимодействуют через события.

from dataclasses import dataclass
from typing import List

# Определение событий
@dataclass
class UserCreatedEvent:
    user_id: int
    name: str
    email: str

@dataclass
class UserEmailUpdatedEvent:
    user_id: int
    new_email: str

# Event Bus
class EventBus:
    def __init__(self):
        self.subscribers: dict = {}
    
    def subscribe(self, event_type, handler):
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(handler)
    
    def publish(self, event):
        event_type = type(event)
        if event_type in self.subscribers:
            for handler in self.subscribers[event_type]:
                handler(event)

# Компоненты, слушающие события
class EmailNotificationService:
    def __init__(self, event_bus: EventBus):
        event_bus.subscribe(UserCreatedEvent, self.on_user_created)
    
    def on_user_created(self, event: UserCreatedEvent):
        print(f"Sending welcome email to {event.email}")

class UserAuditService:
    def __init__(self, event_bus: EventBus):
        event_bus.subscribe(UserCreatedEvent, self.on_user_created)
        event_bus.subscribe(UserEmailUpdatedEvent, self.on_email_updated)
    
    def on_user_created(self, event: UserCreatedEvent):
        print(f"Audit: User {event.user_id} created")
    
    def on_email_updated(self, event: UserEmailUpdatedEvent):
        print(f"Audit: User {event.user_id} email updated")

# Использование
event_bus = EventBus()
EmailNotificationService(event_bus)
UserAuditService(event_bus)

event = UserCreatedEvent(user_id=1, name="Alice", email="alice@example.com")
event_bus.publish(event)

Плюсы: слабая связанность, реактивность, масштабируемость.

Минусы: сложность отладки, консистентность, асинхронность.

6. CQRS (Command Query Responsibility Segregation)

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

# Commands - операции записи
class CreateUserCommand:
    def __init__(self, name: str, email: str):
        self.name = name
        self.email = email

class UpdateUserEmailCommand:
    def __init__(self, user_id: int, new_email: str):
        self.user_id = user_id
        self.new_email = new_email

# Handlers для команд
class CreateUserCommandHandler:
    def __init__(self, repository):
        self.repository = repository
    
    def handle(self, command: CreateUserCommand):
        user = User(name=command.name, email=command.email)
        return self.repository.save(user)

# Queries - операции чтения
class GetUserQuery:
    def __init__(self, user_id: int):
        self.user_id = user_id

# Handlers для запросов
class GetUserQueryHandler:
    def __init__(self, read_repository):
        self.read_repository = read_repository
    
    def handle(self, query: GetUserQuery):
        return self.read_repository.find_user_dto(query.user_id)

7. Serverless архитектура

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

# AWS Lambda function
import json

def lambda_handler(event, context):
    user_id = event['pathParameters']['user_id']
    user = get_user_from_db(user_id)
    
    return {
        'statusCode': 200,
        'body': json.dumps(user)
    }

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

Выводы

Нет одной "лучшей" архитектуры. Выбор зависит от требований проекта, размера команды, бюджета и временных рамок. Часто используется комбинация нескольких стилей.