Как обеспечить надёжность с точки зрения качества компонентов?

Надёжность компонентов: архитектура и качество

Надёжность — это не просто отсутствие ошибок, это предсказуемость, отказоустойчивость и поддерживаемость. Строится на основе хорошей архитектуры и практик.

1. Слоистая архитектура (Layered Architecture)

# ❌ Плохо: смешанные ответственности
def process_order(order_id):
    # Доступ к БД
    order = db.query(Order).get(order_id)
    
    # Валидация
    if order.total < 0:
        return {'error': 'Invalid amount'}
    
    # Бизнес-логика
    order.status = 'processing'
    if order.total > 1000:
        order.is_premium = True
    
    # Отправка email
    send_email(order.customer.email, f'Order {order_id}')
    
    # Сохранение
    db.commit()
    
    # Логирование
    print(f'Order {order_id} processed')
    
    return {'order_id': order_id, 'status': 'processing'}

# ✅ Хорошо: слои с четкими границами
# presentation/ → application/ → domain/ → infrastructure/

Presentation Layer (API, CLI)
    ↓ (вызывает)
Application Layer (Use Cases)
    ↓ (использует)
Domain Layer (Business Logic, Entities)
    ↓ (использует)
Infrastructure Layer (DB, Email, Cache)

⚠️ Зависимости ТОЛЬКО вниз, никогда вверх!

Структура проекта

app/
├── domain/                 # Бизнес-логика, независима от всего
│   ├── entities/
│   │   └── order.py       # Order бизнес-сущность
│   ├── value_objects/
│   │   └── money.py       # Money VO
│   ├── services/          # Domain Services
│   │   └── order_service.py
│   └── exceptions/
│       └── order_exceptions.py
├── application/            # Use Cases, оркестрация
│   ├── use_cases/
│   │   └── process_order_use_case.py
│   └── dtos/
│       └── order_dto.py
├── infrastructure/         # Реализация (БД, email и т.д.)
│   ├── repositories/
│   │   └── order_repository.py
│   ├── services/
│   │   └── email_service.py
│   └── adapters/
│       └── payment_adapter.py
└── presentation/          # API endpoints, CLI
    └── routes/
        └── orders.py

Пример: Order Processing

# app/domain/entities/order.py
from dataclasses import dataclass
from decimal import Decimal
from datetime import datetime
from enum import Enum

class OrderStatus(Enum):
    PENDING = 'pending'
    PROCESSING = 'processing'
    COMPLETED = 'completed'
    FAILED = 'failed'

@dataclass
class Order:
    """Бизнес-сущность Order (независима от инфраструктуры)"""
    id: int
    customer_id: int
    amount: Decimal
    status: OrderStatus
    created_at: datetime
    
    def can_be_processed(self) -> bool:
        """Бизнес-правило: может ли заказ быть обработан"""
        return self.status == OrderStatus.PENDING and self.amount > 0
    
    def mark_as_processing(self):
        """Переход в статус processing"""
        if not self.can_be_processed():
            raise ValueError('Order cannot be processed')
        self.status = OrderStatus.PROCESSING
    
    def mark_as_completed(self):
        """Переход в статус completed"""
        self.status = OrderStatus.COMPLETED

# app/domain/services/order_service.py
class OrderDomainService:
    """Domain service для бизнес-логики заказов"""
    
    def is_premium_order(self, order: Order) -> bool:
        """Бизнес-правило: это премиум заказ?"""
        return order.amount >= Decimal('1000')
    
    def calculate_processing_time(self, order: Order) -> int:
        """Сколько часов обрабатывать заказ"""
        if self.is_premium_order(order):
            return 1  # Премиум обрабатывается быстрее
        return 24

# app/application/use_cases/process_order_use_case.py
from app.domain.entities import Order, OrderStatus
from app.infrastructure.repositories import OrderRepository
from app.infrastructure.services import EmailService, PaymentService
from app.domain.services import OrderDomainService
import logging

logger = logging.getLogger(__name__)

class ProcessOrderUseCase:
    """Use Case: обработать заказ"""
    
    def __init__(
        self,
        order_repository: OrderRepository,
        email_service: EmailService,
        payment_service: PaymentService
    ):
        self.order_repository = order_repository
        self.email_service = email_service
        self.payment_service = payment_service
        self.domain_service = OrderDomainService()
    
    def execute(self, order_id: int) -> dict:
        """Выполнить обработку заказа"""
        try:
            # 1. Получить заказ
            order = self.order_repository.get(order_id)
            if not order:
                raise ValueError(f'Order {order_id} not found')
            
            # 2. Валидация бизнес-правил (domain layer)
            if not order.can_be_processed():
                raise ValueError(f'Order cannot be processed. Status: {order.status}')
            
            # 3. Изменение состояния (domain layer)
            order.mark_as_processing()
            self.order_repository.update(order)
            
            # 4. Отправка письма
            is_premium = self.domain_service.is_premium_order(order)
            self.email_service.send_processing_notification(
                order.customer_id,
                order.id,
                is_premium
            )
            
            # 5. Обработка платежа
            try:
                self.payment_service.process_payment(order.id, order.amount)
            except Exception as e:
                logger.error(f'Payment failed for order {order_id}: {e}')
                order.status = OrderStatus.FAILED
                self.order_repository.update(order)
                raise
            
            # 6. Завершение
            order.mark_as_completed()
            self.order_repository.update(order)
            
            logger.info(f'Order {order_id} processed successfully')
            
            return {
                'order_id': order_id,
                'status': order.status.value,
                'amount': str(order.amount)
            }
        
        except ValueError as e:
            logger.warning(f'Validation error for order {order_id}: {e}')
            raise
        except Exception as e:
            logger.error(f'Unexpected error processing order {order_id}: {e}')
            raise

# app/presentation/routes/orders.py
from fastapi import APIRouter, HTTPException
from app.application.use_cases import ProcessOrderUseCase
from app.infrastructure.repositories import OrderRepository
from app.infrastructure.services import EmailService, PaymentService

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

@router.post('/{order_id}/process')
async def process_order(order_id: int):
    """API endpoint для обработки заказа"""
    try:
        # Инъекция зависимостей
        use_case = ProcessOrderUseCase(
            order_repository=OrderRepository(db_session),
            email_service=EmailService(),
            payment_service=PaymentService()
        )
        
        # Выполнение use case
        result = use_case.execute(order_id)
        return result
    
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail='Internal server error')

2. SOLID принципы

Single Responsibility Principle (SRP)

# ❌ Плохо: класс делает слишком много
class UserService:
    def register(self, email, password):
        # Валидация
        if not email:
            raise ValueError('Email required')
        # Сохранение в БД
        user = User(email=email, password=hash_password(password))
        db.add(user)
        # Отправка email
        send_email(email, 'Welcome')
        # Логирование
        log(f'User {email} registered')
        return user

# ✅ Хорошо: каждый класс одну ответственность
class EmailValidator:
    def validate(self, email: str) -> bool:
        return '@' in email and '.' in email

class PasswordHasher:
    def hash(self, password: str) -> str:
        return bcrypt.hashpw(password.encode(), bcrypt.gensalt())

class UserRepository:
    def create(self, email: str, password_hash: str) -> User:
        user = User(email=email, password_hash=password_hash)
        self.session.add(user)
        self.session.commit()
        return user

class EmailService:
    def send_welcome(self, email: str):
        # Отправка письма
        pass

class UserRegistrationUseCase:
    def __init__(self, validator, hasher, repository, email_service, logger):
        self.validator = validator
        self.hasher = hasher
        self.repository = repository
        self.email_service = email_service
        self.logger = logger
    
    def register(self, email: str, password: str) -> User:
        if not self.validator.validate(email):
            raise ValueError('Invalid email')
        
        password_hash = self.hasher.hash(password)
        user = self.repository.create(email, password_hash)
        
        try:
            self.email_service.send_welcome(email)
        except Exception as e:
            self.logger.error(f'Failed to send welcome email: {e}')
        
        self.logger.info(f'User {email} registered')
        return user

Open/Closed Principle (OCP)

# ❌ Плохо: нужно менять класс при добавлении новой БД
class UserRepository:
    def save(self, user):
        if config.db_type == 'postgresql':
            # PostgreSQL code
            pass
        elif config.db_type == 'mysql':
            # MySQL code
            pass
        elif config.db_type == 'mongodb':
            # MongoDB code
            pass

# ✅ Хорошо: используй абстракцию
from abc import ABC, abstractmethod

class UserRepository(ABC):
    @abstractmethod
    def save(self, user) -> None:
        pass

class PostgreSQLUserRepository(UserRepository):
    def save(self, user):
        # PostgreSQL реализация
        pass

class MongoDBUserRepository(UserRepository):
    def save(self, user):
        # MongoDB реализация
        pass

# Использование (открыто для расширения, закрыто для изменения)
def create_repository() -> UserRepository:
    if config.db_type == 'postgresql':
        return PostgreSQLUserRepository()
    elif config.db_type == 'mongodb':
        return MongoDBUserRepository()

3. Обработка ошибок

# ❌ Плохо: логируешь и бросаешь исключение (double logging)
def process_order(order_id):
    try:
        result = get_order(order_id)
    except OrderNotFound as e:
        logger.error(f'Order not found: {e}')  # Логирование в service
        raise

# На уровне handler
@router.get('/orders/{order_id}')
async def get_order_handler(order_id: int):
    try:
        return process_order(order_id)
    except OrderNotFound as e:
        logger.error(f'Order not found: {e}')  # ЕЩЕ раз логирование!
        raise HTTPException(status_code=404)

# ✅ Правильно: логируй один раз на границе
class OrderRepository:
    def get(self, order_id: int) -> Order:
        """Не логируем — просто бросаем исключение"""
        order = self.session.query(Order).get(order_id)
        if not order:
            raise OrderNotFound(order_id)
        return order

# На уровне handler (граница системы)
@router.get('/orders/{order_id}')
async def get_order_handler(order_id: int):
    try:
        return get_order_service.execute(order_id)
    except OrderNotFound as e:
        logger.warning(f'Order not found: {e}')  # Логируем один раз
        raise HTTPException(status_code=404)
    except Exception as e:
        logger.error(f'Unexpected error: {e}')  # Перехватываем неожиданное
        raise HTTPException(status_code=500)

4. Тестирование

# tests/unit/application/test_process_order_use_case.py
import pytest
from unittest.mock import Mock
from app.application.use_cases import ProcessOrderUseCase
from app.domain.entities import Order, OrderStatus

class TestProcessOrderUseCase:
    @pytest.fixture
    def mock_dependencies(self):
        return {
            'order_repository': Mock(),
            'email_service': Mock(),
            'payment_service': Mock()
        }
    
    @pytest.fixture
    def use_case(self, mock_dependencies):
        return ProcessOrderUseCase(**mock_dependencies)
    
    def test_process_successful_order(self, use_case, mock_dependencies):
        # Arrange
        order = Order(
            id=1,
            customer_id=1,
            amount=Decimal('100'),
            status=OrderStatus.PENDING,
            created_at=datetime.now()
        )
        mock_dependencies['order_repository'].get.return_value = order
        
        # Act
        result = use_case.execute(1)
        
        # Assert
        assert result['status'] == 'completed'
        mock_dependencies['payment_service'].process_payment.assert_called_once()
        mock_dependencies['email_service'].send_processing_notification.assert_called_once()

5. Метрики качества

# Test Coverage
make coverage
# Цель: > 90%

# Code Quality (Linting)
make lint
# Используй ruff, pylint, flake8

# Type Checking
mypy app/
# Строгая типизация

# Complexity
radon cc app/ -a
# Циклическая сложность < 10

# Code Duplication
radon mi app/ -m
# Maintainability Index > 80

6. Лучшие практики

1. Слои архитектуры — четкие границы
2. SOLID принципы — гибкий и масштабируемый код
3. DRY, KISS — простота и переиспользование
4. Тестирование — 90%+ coverage
5. Обработка ошибок — логируй один раз на границе
6. Dependency Injection — легко мокировать
7. Code Review — проверка качества
8. Документация — объясни сложные части

Надёжность строится на фундаменте хорошей архитектуры и дисциплины!