Как создается сервис?

Как создается сервис (Service в Clean Architecture)

Сервис — это ключевой компонент в архитектуре приложения. Вот полное руководство по создании.

1. Что такое сервис?

Сервис — это класс, который:

• Содержит бизнес-логику приложения
• Взаимодействует с репозиториями для доступа к данным
• Не зависит от деталей реализации (БД, API, UI)
• Можно тестировать в изоляции
• Используется контроллерами/представлениями

Пирамида архитектуры (от внешнего к внутреннему):

    ┌─────────────────────────┐
    │ Presentation (UI/API)   │ ← REST endpoints, web pages
    ├─────────────────────────┤
    │ Application (Services)  │ ← Бизнес-логика ← ЭТО ТО, ЧТО НАМ НУЖНО
    ├─────────────────────────┤
    │ Domain (Models)         │ ← Сущности, правила
    ├─────────────────────────┤
    │ Infrastructure          │ ← БД, HTTP клиенты, файлы
    └─────────────────────────┘

2. Простой пример: UserService

# models.py (Domain layer)
from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    """Модель пользователя (domain entity)"""
    id: int
    email: str
    username: str
    password_hash: str
    created_at: datetime
    is_active: bool

# repositories.py (Infrastructure layer)
from abc import ABC, abstractmethod
from typing import Optional

class UserRepository(ABC):
    """Интерфейс репозитория для работы с пользователями"""
    
    @abstractmethod
    def find_by_email(self, email: str) -> Optional[User]:
        """Найти пользователя по email"""
        pass
    
    @abstractmethod
    def find_by_id(self, user_id: int) -> Optional[User]:
        """Найти пользователя по ID"""
        pass
    
    @abstractmethod
    def create(self, user: User) -> User:
        """Создать новго пользователя"""
        pass
    
    @abstractmethod
    def update(self, user: User) -> User:
        """Обновить пользователя"""
        pass

class DatabaseUserRepository(UserRepository):
    """Реальная реализация через БД"""
    
    def __init__(self, db_session):
        self.db = db_session
    
    def find_by_email(self, email: str) -> Optional[User]:
        user_model = self.db.query(UserModel).filter_by(email=email).first()
        return self._to_domain(user_model) if user_model else None
    
    def find_by_id(self, user_id: int) -> Optional[User]:
        user_model = self.db.query(UserModel).filter_by(id=user_id).first()
        return self._to_domain(user_model) if user_model else None
    
    def create(self, user: User) -> User:
        user_model = UserModel(
            email=user.email,
            username=user.username,
            password_hash=user.password_hash,
            is_active=user.is_active
        )
        self.db.add(user_model)
        self.db.commit()
        return self._to_domain(user_model)
    
    def update(self, user: User) -> User:
        user_model = self.db.query(UserModel).get(user.id)
        user_model.username = user.username
        user_model.is_active = user.is_active
        self.db.commit()
        return self._to_domain(user_model)
    
    @staticmethod
    def _to_domain(user_model) -> User:
        return User(
            id=user_model.id,
            email=user_model.email,
            username=user_model.username,
            password_hash=user_model.password_hash,
            created_at=user_model.created_at,
            is_active=user_model.is_active
        )

# services.py (Application layer) ← ГЛАВНЫЙ СЛОЙ
from typing import Optional
from passlib.context import CryptContext

class UserService:
    """Сервис для работы с пользователями"""
    
    def __init__(self, user_repo: UserRepository):
        """Инъекция зависимости"""
        self.repo = user_repo
        self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
    
    def register_user(self, email: str, username: str, password: str) -> User:
        """Зарегистрировать нового пользователя"""
        
        # Проверяем что email не занят
        if self.repo.find_by_email(email):
            raise ValueError(f"Email {email} already registered")
        
        # Хешируем пароль
        password_hash = self.pwd_context.hash(password)
        
        # Создаем пользователя
        user = User(
            id=None,
            email=email,
            username=username,
            password_hash=password_hash,
            created_at=datetime.now(),
            is_active=True
        )
        
        # Сохраняем через репозиторий
        return self.repo.create(user)
    
    def authenticate(self, email: str, password: str) -> Optional[User]:
        """Проверить email и пароль"""
        user = self.repo.find_by_email(email)
        
        if not user:
            return None
        
        # Проверяем пароль
        if self.pwd_context.verify(password, user.password_hash):
            return user
        
        return None
    
    def get_user_by_id(self, user_id: int) -> Optional[User]:
        """Получить пользователя по ID"""
        return self.repo.find_by_id(user_id)
    
    def deactivate_user(self, user_id: int) -> User:
        """Деактивировать пользователя"""
        user = self.repo.find_by_id(user_id)
        
        if not user:
            raise ValueError(f"User {user_id} not found")
        
        user.is_active = False
        return self.repo.update(user)

3. Использование сервиса в контроллере

# controllers/user_controller.py (Presentation layer)
from flask import request, jsonify
from services import UserService

class UserController:
    def __init__(self, user_service: UserService):
        self.service = user_service
    
    def register(self):
        """POST /users/register"""
        data = request.json
        
        try:
            user = self.service.register_user(
                email=data['email'],
                username=data['username'],
                password=data['password']
            )
            return jsonify({'id': user.id, 'email': user.email}), 201
        except ValueError as e:
            return jsonify({'error': str(e)}), 400
    
    def login(self):
        """POST /users/login"""
        data = request.json
        
        user = self.service.authenticate(
            email=data['email'],
            password=data['password']
        )
        
        if user:
            return jsonify({'id': user.id, 'email': user.email}), 200
        else:
            return jsonify({'error': 'Invalid credentials'}), 401
    
    def get_profile(self, user_id: int):
        """GET /users/{user_id}"""
        user = self.service.get_user_by_id(user_id)
        
        if not user:
            return jsonify({'error': 'User not found'}), 404
        
        return jsonify({
            'id': user.id,
            'email': user.email,
            'username': user.username,
            'is_active': user.is_active
        }), 200

4. Инъекция зависимостей

# main.py (Composition root)
from flask import Flask
from database import get_db_session
from repositories import DatabaseUserRepository
from services import UserService
from controllers import UserController

app = Flask(__name__)

# Создаем зависимости
db_session = get_db_session()
user_repo = DatabaseUserRepository(db_session)
user_service = UserService(user_repo)
user_controller = UserController(user_service)

# Регистрируем маршруты
@app.route('/users/register', methods=['POST'])
def register():
    return user_controller.register()

@app.route('/users/login', methods=['POST'])
def login():
    return user_controller.login()

@app.route('/users/<int:user_id>', methods=['GET'])
def get_profile(user_id):
    return user_controller.get_profile(user_id)

if __name__ == '__main__':
    app.run()

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

# tests/test_user_service.py
import pytest
from unittest.mock import Mock, MagicMock
from services import UserService
from models import User
from datetime import datetime

class MockUserRepository:
    """Мок репозитория для тестирования"""
    
    def __init__(self):
        self.users = {}
        self.next_id = 1
    
    def find_by_email(self, email):
        for user in self.users.values():
            if user.email == email:
                return user
        return None
    
    def find_by_id(self, user_id):
        return self.users.get(user_id)
    
    def create(self, user):
        user.id = self.next_id
        self.users[self.next_id] = user
        self.next_id += 1
        return user
    
    def update(self, user):
        self.users[user.id] = user
        return user

def test_register_user_success():
    """Тест: успешная регистрация"""
    # Arrange
    repo = MockUserRepository()
    service = UserService(repo)
    
    # Act
    user = service.register_user('alice@example.com', 'alice', 'password123')
    
    # Assert
    assert user.id == 1
    assert user.email == 'alice@example.com'
    assert user.username == 'alice'
    assert user.is_active == True

def test_register_duplicate_email():
    """Тест: дублирование email"""
    repo = MockUserRepository()
    service = UserService(repo)
    
    # Регистрируем первого пользователя
    service.register_user('alice@example.com', 'alice', 'password123')
    
    # Пытаемся зарегистрировать с тем же email
    with pytest.raises(ValueError, match="already registered"):
        service.register_user('alice@example.com', 'alice2', 'password456')

def test_authenticate_success():
    """Тест: успешная аутентификация"""
    repo = MockUserRepository()
    service = UserService(repo)
    
    # Регистрируем пользователя
    service.register_user('alice@example.com', 'alice', 'password123')
    
    # Аутентифицируемся
    user = service.authenticate('alice@example.com', 'password123')
    
    assert user is not None
    assert user.email == 'alice@example.com'

def test_authenticate_wrong_password():
    """Тест: неверный пароль"""
    repo = MockUserRepository()
    service = UserService(repo)
    
    service.register_user('alice@example.com', 'alice', 'password123')
    
    user = service.authenticate('alice@example.com', 'wrongpassword')
    
    assert user is None

def test_authenticate_nonexistent_user():
    """Тест: несуществующий пользователь"""
    repo = MockUserRepository()
    service = UserService(repo)
    
    user = service.authenticate('nonexistent@example.com', 'password')
    
    assert user is None

6. Сложный пример: OrderService

# models.py
@dataclass
class Order:
    id: int
    user_id: int
    items: list  # [{'product_id': 1, 'quantity': 2}]
    total_amount: float
    status: str  # 'pending', 'paid', 'shipped', 'delivered'
    created_at: datetime

# services.py
class OrderService:
    def __init__(
        self,
        order_repo,
        product_repo,
        payment_service,  # Внешний сервис
        email_service     # Для отправки уведомлений
    ):
        self.order_repo = order_repo
        self.product_repo = product_repo
        self.payment_service = payment_service
        self.email_service = email_service
    
    def create_order(self, user_id: int, items: list) -> Order:
        """Создать заказ с проверками"""
        
        # 1. Валидация
        if not items:
            raise ValueError("Order must have at least one item")
        
        # 2. Проверить наличие товаров
        total_amount = 0
        order_items = []
        
        for item in items:
            product = self.product_repo.find_by_id(item['product_id'])
            if not product:
                raise ValueError(f"Product {item['product_id']} not found")
            
            if product.stock < item['quantity']:
                raise ValueError(f"Insufficient stock for {product.name}")
            
            total_amount += product.price * item['quantity']
            order_items.append(item)
        
        # 3. Создать заказ
        order = Order(
            id=None,
            user_id=user_id,
            items=order_items,
            total_amount=total_amount,
            status='pending',
            created_at=datetime.now()
        )
        
        order = self.order_repo.create(order)
        
        # 4. Отправить email
        self.email_service.send_order_confirmation(user_id, order)
        
        return order
    
    def pay_for_order(self, order_id: int, payment_token: str) -> bool:
        """Обработать платеж"""
        order = self.order_repo.find_by_id(order_id)
        
        if not order:
            raise ValueError(f"Order {order_id} not found")
        
        if order.status != 'pending':
            raise ValueError(f"Order already {order.status}")
        
        try:
            # Проверить платеж через внешний сервис
            payment_result = self.payment_service.process_payment(
                token=payment_token,
                amount=order.total_amount
            )
            
            if not payment_result['success']:
                return False
            
            # Обновить статус заказа
            order.status = 'paid'
            self.order_repo.update(order)
            
            # Отправить уведомление
            self.email_service.send_payment_confirmation(order.user_id, order)
            
            return True
        
        except Exception as e:
            logger.error(f"Payment error for order {order_id}: {e}")
            raise

7. Правила создания сервиса

# ✅ ХОРОШО: Одна ответственность
class UserService:
    """Только логика работы с пользователями"""
    def register_user(self, ...): pass
    def authenticate(self, ...): pass
    def get_user(self, ...): pass

# ❌ ПЛОХО: Смешивание ответственности
class UserService:
    """Слишком много разных функций"""
    def register_user(self, ...): pass
    def send_email(self, ...): pass  # Это должно быть в отдельном сервисе
    def save_to_cache(self, ...): pass  # Это тоже
    def log_activity(self, ...): pass  # И это

# ✅ ХОРОШО: Использовать интерфейсы
class UserService:
    def __init__(self, user_repo: UserRepository):  # Интерфейс!
        self.repo = user_repo

# ❌ ПЛОХО: Жесткие зависимости
class UserService:
    def __init__(self):
        self.repo = DatabaseUserRepository()  # Сложно тестировать

# ✅ ХОРОШО: Валидация в сервисе
class UserService:
    def register_user(self, email, password):
        if '@' not in email:
            raise ValueError("Invalid email")
        if len(password) < 8:
            raise ValueError("Password too short")

# ✅ ХОРОШО: Использовать domain exceptions
class EmailAlreadyRegisteredException(Exception):
    pass

class UserService:
    def register_user(self, email, password):
        if self.repo.find_by_email(email):
            raise EmailAlreadyRegisteredException(email)

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

project/
├── app.py                          # Entry point
├── requirements.txt
│
├── domain/                         # Ядро - бизнес-логика
│   ├── __init__.py
│   ├── models.py                   # Entities, Value Objects
│   ├── exceptions.py               # Domain exceptions
│   └── repositories.py             # Интерфейсы репозиториев
│
├── application/                    # Бизнес-процессы
│   ├── __init__.py
│   ├── services/
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   ├── order_service.py
│   │   └── payment_service.py
│   ├── dto.py                      # Data Transfer Objects
│   └── use_cases.py                # Use Cases
│
├── infrastructure/                 # Реализации
│   ├── __init__.py
│   ├── repositories/
│   │   ├── __init__.py
│   │   ├── user_repository.py
│   │   └── order_repository.py
│   ├── clients/
│   │   ├── payment_client.py
│   │   └── email_client.py
│   └── database.py
│
├── presentation/                   # API/Web
│   ├── __init__.py
│   ├── controllers/
│   │   ├── __init__.py
│   │   ├── user_controller.py
│   │   └── order_controller.py
│   └── schemas.py
│
└── tests/
    ├── test_user_service.py
    ├── test_order_service.py
    └── conftest.py

Заключение

Сервис в Clean Architecture:

1. Содержит бизнес-логику (главная ценность приложения)
2. Не зависит от деталей реализации (БД, API, фреймворк)
3. Легко тестировать (благодаря инъекции зависимостей)
4. Одна ответственность (Single Responsibility Principle)
5. Использует интерфейсы (Dependency Inversion Principle)

Это делает код:

• Более понятным
• Легче поддерживаемым
• Проще тестируемым
• Более модульным и переиспользуемым