Чистая архитектура в Python: подробное руководство с примерами

Чистая архитектура (Clean Architecture) — это концепция проектирования программного обеспечения, предложенная Робертом Мартином (дядя Боб). Её основная цель — создание систем, которые легко поддерживать, тестировать и модифицировать. В этой статье мы подробно разберём принципы чистой архитектуры и покажем, как применять их в Python-проектах.

Основные принципы чистой архитектуры

Зависимости направлены внутрь

Самый важный принцип чистой архитектуры — правило зависимостей: зависимости в коде должны быть направлены внутрь, к центру системы, где находится бизнес-логика. Внешние слои (например, база данных или веб-интерфейс) зависят от внутренних слоёв, но не наоборот.

Независимость от фреймворков

Бизнес-логика не должна зависеть от внешних фреймворков (Django, Flask, SQLAlchemy и т.д.). Это позволяет легко менять технологии без переписывания основной логики приложения.

Тестируемость

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

Слои чистой архитектуры

1. Entities (Сущности) — основные бизнес-объекты

2. Use Cases (Сценарии использования) — бизнес-правила

3. Interface Adapters (Адаптеры интерфейсов) — преобразователи данных

4. Frameworks & Drivers (Фреймворки и драйверы) — внешние компоненты

Реализация чистой архитектуры в Python

Рассмотрим реализацию на примере системы управления задачами.

1. Сущности (Entities)

Сущности содержат основные бизнес-объекты и правила, которые не зависят от конкретной реализации.

from dataclasses import dataclass

from datetime import datetime

from typing import Optional

@dataclass

class Task:

....id: Optional[int]

....title: str

....description: str

....completed: bool

....created_at: datetime

....updated_at: datetime

....def __post_init__(self):

........if self.id is None:

........self.id = id(self)

........if self.created_at is None:

............self.created_at = datetime.now()

........if self.updated_at is None:

............self.updated_at = datetime.now()

....def mark_completed(self):

........self.completed = True

........self.updated_at = datetime.now()

....def update_title(self, new_title: str):

........if not new_title.strip():

............raise ValueError("Title cannot be empty")

........self.title = new_title

........self.updated_at = datetime.now()

2. Репозитории (Repositories)

Репозитории определяют интерфейсы для работы с данными, но не их реализацию.

from abc import ABC, abstractmethod

from typing import List, Optional

class TaskRepository(ABC):

....@abstractmethod

....def get_by_id(self, task_id: int) -> Optional[Task]:

........pass

....@abstractmethod

........def get_all(self) -> List[Task]:

............pass

....@abstractmethod

....def add(self, task: Task) -> Task:

........pass

....@abstractmethod

....def update(self, task: Task) -> Task:

........pass

....@abstractmethod

....def delete(self, task_id: int) -> bool:

........pass

3. Сценарии использования (Use Cases)

Сценарии использования содержат бизнес-логику приложения.

class TaskService:

....def __init__(self, task_repository: TaskRepository):

........self.task_repository = task_repository

....def create_task(self, title: str, description: str) -> Task:

........if not title.strip():

............raise ValueError("Title cannot be empty")

........task = Task(

............id=None,

............title=title,

............description=description,

............completed=False,

............created_at=None,

............updated_at=None

........)

....return self.task_repository.add(task)

....def complete_task(self, task_id: int) -> Optional[Task]:

........task = self.task_repository.get_by_id(task_id)

........if task:

............task.mark_completed()

............return self.task_repository.update(task)

........return None

....def get_pending_tasks(self) -> List[Task]:

........all_tasks = self.task_repository.get_all()

........return [task for task in all_tasks if not task.completed]

....def get_completed_tasks(self) -> List[Task]:

........all_tasks = self.task_repository.get_all()

........return [task for task in all_tasks if task.completed]

4. Адаптеры (Adapters)

Адаптеры преобразуют данные между слоями.

Репозиторий для базы данных

from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class TaskModel(Base):

....__tablename__ = 'tasks'

....id = Column(Integer, primary_key=True)

....title = Column(String(255), nullable=False)

....description = Column(String(1000))

....completed = Column(Boolean, default=False)

....created_at = Column(DateTime, default=datetime.now)

....updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)

class SQLAlchemyTaskRepository(TaskRepository):

....def __init__(self, session):

........self.session = session

....def get_by_id(self, task_id: int) -> Optional[Task]:

........task_model = self.session.query(TaskModel).filter_by(id=task_id).first()

........if task_model:

............return self._to_entity(task_model)

........return None

....def get_all(self) -> List[Task]:

........task_models = self.session.query(TaskModel).all()

........return [self._to_entity(model) for model in task_models]

....def add(self, task: Task) -> Task:

........task_model = TaskModel(

............title=task.title,

............description=task.description,

............completed=task.completed,

............created_at=task.created_at,

............updated_at=task.updated_at

........)

........self.session.add(task_model)

........self.session.commit()

........return self._to_entity(task_model)

....def update(self, task: Task) -> Task:

........task_model = self.session.query(TaskModel).filter_by(id=task.id).first()

........if task_model:

............task_model.title = task.title

............task_model.description = task.description

............task_model.completed = task.completed

............task_model.updated_at = task.updated_at

............self.session.commit()

............return self._to_entity(task_model)

........raise ValueError(f"Task with id {task.id} not found")

....def delete(self, task_id: int) -> bool:

........task_model = self.session.query(TaskModel).filter_by(id=task_id).first()

........if task_model:

............self.session.delete(task_model)

............self.session.commit()

............return True

........return False

....def _to_entity(self, model: TaskModel) -> Task:

........return Task(

............id=model.id,

............title=model.title,

............description=model.description,

............completed=model.completed,

............created_at=model.created_at,

............updated_at=model.updated_at

........)

....def _to_model(self, entity: Task) -> TaskModel:

........return TaskModel(

............id=entity.id,

............title=entity.title,

............description=entity.description,

............completed=entity.completed,

............created_at=entity.created_at,

............updated_at=entity.updated_at

........)

Репозиторий для работы в памяти (для тестирования)

class InMemoryTaskRepository(TaskRepository):

....def __init__(self):

........self.tasks = {}

........self.next_id = 1

....def get_by_id(self, task_id: int) -> Optional[Task]:

........return self.tasks.get(task_id)

....def get_all(self) -> List[Task]:

........return list(self.tasks.values())

....def add(self, task: Task) -> Task:

........task.id = self.next_id

........self.tasks[task.id] = task

........self.next_id += 1

........return task

....def update(self, task: Task) -> Task:

........if task.id not in self.tasks:

............raise ValueError(f"Task with id {task.id} not found")

........self.tasks[task.id] = task

........return task

....def delete(self, task_id: int) -> bool:

........if task_id in self.tasks:

............del self.tasks[task_id]

............return True

........return False

5. Внешний слой (Frameworks & Drivers)

Веб-интерфейс с Flask

from flask import Flask, request, jsonify

app = Flask(__name__)

# Инициализация зависимостей

def setup_dependencies():

....# В реальном приложении здесь была бы инициализация БД

....repository = InMemoryTaskRepository()

....task_service = TaskService(repository)

....return task_service

task_service = setup_dependencies()

@app.route('/tasks', methods=['GET'])

def get_tasks():

....status = request.args.get('status', 'all')

....if status == 'completed':

........tasks = task_service.get_completed_tasks()

....elif status == 'pending':

........tasks = task_service.get_pending_tasks()

....else:

........tasks = task_service.task_repository.get_all()

....return jsonify([{

........'id': task.id,

........'title': task.title,

........'description': task.description,

........'completed': task.completed,

........'created_at': task.created_at.isoformat(),

........'updated_at': task.updated_at.isoformat()

....} for task in tasks])

@app.route('/tasks', methods=['POST'])

def create_task():

....data = request.get_json()

....try:

........task = task_service.create_task(

............title=data['title'],

............description=data.get('description', '')

........)

........return jsonify({

............'id': task.id,

............'title': task.title,

............'description': task.description,

............'completed': task.completed,

............'created_at': task.created_at.isoformat(),

............'updated_at': task.updated_at.isoformat()

........}), 201

....except ValueError as e:

........return jsonify({'error': str(e)}), 400

@app.route('/tasks/<int:task_id>/complete', methods=['POST'])

def complete_task(task_id):

....task = task_service.complete_task(task_id)

....if task:

........return jsonify({

............'id': task.id,

............'title': task.title,

............'description': task.description,

............'completed': task.completed,

............'created_at': task.created_at.isoformat(),

............'updated_at': task.updated_at.isoformat()

........})

....else:

........return jsonify({'error': 'Task not found'}), 404

if __name__ == '__main__':

....app.run(debug=True)

Консольный интерфейс

class TaskCLI:

....def __init__(self, task_service: TaskService):

........self.task_service = task_service

....def run(self):

........while True:

............print("\n=== Task Management System ===")

............print("1. List all tasks")

............print("2. List pending tasks")

............print("3. List completed tasks")

............print("4. Create new task")

............print("5. Complete task")

............print("6. Exit")

............choice = input("Enter your choice: ")

............if choice == '1':

................self.list_all_tasks()

............elif choice == '2':

................self.list_pending_tasks()

............elif choice == '3':

................self.list_completed_tasks()

............elif choice == '4':

................self.create_task()

............elif choice == '5':

................self.complete_task()

............elif choice == '6':

................break

............else:

................print("Invalid choice. Please try again.")

....def list_all_tasks(self):

........tasks = self.task_service.task_repository.get_all()

........self._print_tasks(tasks)

....def list_pending_tasks(self):

........tasks = self.task_service.get_pending_tasks()

........self._print_tasks(tasks)

....def list_completed_tasks(self):

........tasks = self.task_service.get_completed_tasks()

........self._print_tasks(tasks)

....def create_task(self):

........title = input("Enter task title: ")

........description = input("Enter task description: ")

........try:

............task = self.task_service.create_task(title, description)

............print(f"Task created successfully with ID: {task.id}")

........except ValueError as e:

............print(f"Error: {e}")

....def complete_task(self):

........task_id = int(input("Enter task ID to complete: "))

........task = self.task_service.complete_task(task_id)

........if task:

............print(f"Task {task_id} marked as completed")

........else:

............print(f"Task with ID {task_id} not found")

....def _print_tasks(self, tasks):

........if not tasks:

............print("No tasks found.")

............return

........for task in tasks:

............status = "Completed" if task.completed else "Pending"

............print(f"{task.id}. {task.title} [{status}]")

............print(f" {task.description}")

............print(f" Created: {task.created_at}, Updated: {task.updated_at}")

............print()

# Использование

if __name__ == "__main__":

....repository = InMemoryTaskRepository()

....task_service = TaskService(repository)

....cli = TaskCLI(task_service)

....cli.run()

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

Одно из главных преимуществ чистой архитектуры — простота тестирования.

Тестирование бизнес-логики

import unittest

from datetime import datetime

class TestTaskService(unittest.TestCase):

....def setUp(self):

........self.repository = InMemoryTaskRepository()

........self.task_service = TaskService(self.repository)

....def test_create_task(self):

........task = self.task_service.create_task("Test task", "Test description")

........self.assertEqual(task.title, "Test task")

........self.assertEqual(task.description, "Test description")

........self.assertFalse(task.completed)

........self.assertIsNotNone(task.created_at)

........self.assertIsNotNone(task.updated_at)

....def test_create_task_with_empty_title(self):

........with self.assertRaises(ValueError):

............self.task_service.create_task("", "Test description")

....def test_complete_task(self):

........task = self.task_service.create_task("Test task", "Test description")

........completed_task = self.task_service.complete_task(task.id)

........self.assertTrue(completed_task.completed)

........self.assertGreater(completed_task.updated_at, task.updated_at)

....def test_complete_nonexistent_task(self):

........result = self.task_service.complete_task(999)

........self.assertIsNone(result)

....def test_get_pending_tasks(self):

........task1 = self.task_service.create_task("Task 1", "Description 1")

........task2 = self.task_service.create_task("Task 2", "Description 2")

........self.task_service.complete_task(task1.id)

........pending_tasks = self.task_service.get_pending_tasks()

........self.assertEqual(len(pending_tasks), 1)

........self.assertEqual(pending_tasks[0].id, task2.id)

....def test_get_completed_tasks(self):

........task1 = self.task_service.create_task("Task 1", "Description 1")

........task2 = self.task_service.create_task("Task 2", "Description 2")

........self.task_service.complete_task(task1.id)

........completed_tasks = self.task_service.get_completed_tasks()

........self.assertEqual(len(completed_tasks), 1)

........self.assertEqual(completed_tasks[0].id, task1.id)

if __name__ == '__main__':

....unittest.main()

Интеграционные тесты

class TestTaskServiceWithDatabase(unittest.TestCase):

....def setUp(self):

........# Создание тестовой БД в памяти

........engine = create_engine('sqlite:///:memory:')

........Base.metadata.create_all(engine)

........Session = sessionmaker(bind=engine)

........session = Session()

........self.repository = SQLAlchemyTaskRepository(session)

........self.task_service = TaskService(self.repository)

....def test_create_and_retrieve_task(self):

........task = self.task_service.create_task("Test task", "Test description")

........retrieved_task = self.task_service.task_repository.get_by_id(task.id)

........self.assertEqual(retrieved_task.title, "Test task")

........self.assertEqual(retrieved_task.description, "Test description")

Преимущества чистой архитектуры

1. Легкость тестирования: Бизнес-логика изолирована от внешних зависимостей.

2. Независимость от фреймворков: Можно легко заменить веб-фреймворк или базу данных.

3. Удобство поддержки: Четкое разделение ответственности между компонентами.

4. Гибкость: Легко добавлять новые функции и интерфейсы.

5. Масштабируемость: Компоненты можно развивать независимо друг от друга.

Недостатки и сложности

1. Избыточность для простых проектов: Для небольших приложений чистая архитектура может показаться излишне сложной.

2. Кривая обучения: Требует времени на освоение концепций.

3. Больше кода: Необходимость создавать интерфейсы и адаптеры увеличивает объем кода.

Заключение

Чистая архитектура — это мощный подход к проектированию программного обеспечения, который особенно полезен для сложных, долгосрочных проектов. В Python её реализация требует дисциплины и следования принципам, но результат стоит усилий: получается гибкое, тестируемое и поддерживаемое приложение.

Ключевые моменты для успешной реализации:

1. Строгое разделение слоев

2. Зависимости направлены внутрь

3. Бизнес-логика независима от фреймворков

4. Использование интерфейсов (абстрактных классов) для определения контрактов

Начните с малого, применяйте принципы постепенно, и вы оцените преимущества чистой архитектуры в ваших Python-проектах.

Подписывайтесь:

Телеграм https://t.me/lets_go_code
Канал "Просто о программировании" https://dzen.ru/lets_go_code