Python Разработчик-Эксперт

🐍 Профессиональное Python Решение

from typing import List, Dict, Optional, Protocol
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from abc import ABC, abstractmethod
import asyncio
import pytest

@dataclass
class Task:
    """Модель задачи с полной типизацией."""
    id: int
    title: str
    description: Optional[str] = None
    priority: int = field(default=1)
    created_at: datetime = field(default_factory=datetime.now)
    due_date: Optional[datetime] = None
    completed: bool = False
    tags: List[str] = field(default_factory=list)
    
    def __post_init__(self):
        """Валидация после инициализации."""
        if not 1 <= self.priority <= 10:
            raise ValueError("Приоритет должен быть от 1 до 10")
        
        if self.due_date and self.due_date <= self.created_at:
            raise ValueError("Срок выполнения не может быть в прошлом")

class TaskFilter(Protocol):
    """Протокол для фильтрации задач."""
    
    def apply(self, tasks: List[Task]) -> List[Task]:
        """Применить фильтр к списку задач."""
        ...

class PriorityFilter:
    """Фильтр по приоритету задач."""
    
    def __init__(self, min_priority: int):
        self.min_priority = min_priority
    
    def apply(self, tasks: List[Task]) -> List[Task]:
        return [task for task in tasks if task.priority >= self.min_priority]

class TaskRepository(ABC):
    """Абстрактный репозиторий для работы с задачами."""
    
    @abstractmethod
    async def save(self, task: Task) -> Task:
        """Сохранить задачу."""
        pass
    
    @abstractmethod
    async def find_by_id(self, task_id: int) -> Optional[Task]:
        """Найти задачу по ID."""
        pass
    
    @abstractmethod
    async def find_all(self) -> List[Task]:
        """Получить все задачи."""
        pass

class InMemoryTaskRepository(TaskRepository):
    """In-memory реализация репозитория для тестирования."""
    
    def __init__(self):
        self._tasks: Dict[int, Task] = {}
        self._next_id = 1
    
    async def save(self, task: Task) -> Task:
        """Сохранить задачу в памяти."""
        if task.id == 0:  # Новая задача
            task.id = self._next_id
            self._next_id += 1
        
        self._tasks[task.id] = task
        return task
    
    async def find_by_id(self, task_id: int) -> Optional[Task]:
        """Найти задачу по ID."""
        return self._tasks.get(task_id)
    
    async def find_all(self) -> List[Task]:
        """Получить все задачи."""
        return list(self._tasks.values())

class TaskService:
    """Сервис для управления задачами."""
    
    def __init__(self, repository: TaskRepository):
        self._repository = repository
    
    async def create_task(
        self,
        title: str,
        description: Optional[str] = None,
        priority: int = 1,
        due_date: Optional[datetime] = None,
        tags: Optional[List[str]] = None
    ) -> Task:
        """
        Создать новую задачу.
        
        Args:
            title: Название задачи
            description: Описание задачи
            priority: Приоритет от 1 до 10
            due_date: Срок выполнения
            tags: Список тегов
            
        Returns:
            Созданная задача
            
        Raises:
            ValueError: При некорректных параметрах
        """
        if not title.strip():
            raise ValueError("Название задачи не может быть пустым")
        
        task = Task(
            id=0,  # Будет назначен в репозитории
            title=title.strip(),
            description=description,
            priority=priority,
            due_date=due_date,
            tags=tags or []
        )
        
        return await self._repository.save(task)
    
    async def complete_task(self, task_id: int) -> Optional[Task]:
        """Отметить задачу как выполненную."""
        task = await self._repository.find_by_id(task_id)
        if not task:
            return None
        
        task.completed = True
        return await self._repository.save(task)
    
    async def get_filtered_tasks(
        self, 
        task_filter: TaskFilter
    ) -> List[Task]:
        """Получить отфильтрованные задачи."""
        all_tasks = await self._repository.find_all()
        return task_filter.apply(all_tasks)
    
    async def get_overdue_tasks(self) -> List[Task]:
        """Получить просроченные задачи."""
        all_tasks = await self._repository.find_all()
        now = datetime.now()
        
        return [
            task for task in all_tasks 
            if task.due_date and task.due_date < now and not task.completed
        ]

# Unit Tests
class TestTaskService:
    """Тесты для TaskService."""
    
    @pytest.fixture
    async def service(self):
        """Создать сервис с in-memory репозиторием."""
        repository = InMemoryTaskRepository()
        return TaskService(repository)
    
    @pytest.mark.asyncio
    async def test_create_task_success(self, service):
        """Тест успешного создания задачи."""
        task = await service.create_task(
            title="Тестовая задача",
            description="Описание задачи",
            priority=5
        )
        
        assert task.id > 0
        assert task.title == "Тестовая задача"
        assert task.priority == 5
        assert not task.completed
    
    @pytest.mark.asyncio
    async def test_create_task_empty_title(self, service):
        """Тест создания задачи с пустым названием."""
        with pytest.raises(ValueError, match="Название задачи не может быть пустым"):
            await service.create_task(title="   ")
    
    @pytest.mark.asyncio
    async def test_complete_task(self, service):
        """Тест завершения задачи."""
        # Создаем задачу
        task = await service.create_task("Задача для завершения")
        
        # Завершаем её
        completed_task = await service.complete_task(task.id)
        
        assert completed_task is not None
        assert completed_task.completed is True
    
    @pytest.mark.asyncio
    async def test_filter_by_priority(self, service):
        """Тест фильтрации по приоритету."""
        # Создаем задачи с разными приоритетами
        await service.create_task("Низкий приоритет", priority=2)
        await service.create_task("Высокий приоритет", priority=8)
        
        # Фильтруем задачи с приоритетом >= 5
        high_priority_filter = PriorityFilter(5)
        filtered_tasks = await service.get_filtered_tasks(high_priority_filter)
        
        assert len(filtered_tasks) == 1
        assert filtered_tasks[0].title == "Высокий приоритет"

# Пример использования
async def main():
    """Демонстрация работы системы управления задачами."""
    # Создание сервиса
    repository = InMemoryTaskRepository()
    service = TaskService(repository)
    
    # Создание задач
    task1 = await service.create_task(
        title="Изучить FastAPI",
        description="Освоить создание REST API",
        priority=8,
        due_date=datetime.now() + timedelta(days=7),
        tags=["обучение", "python"]
    )
    
    task2 = await service.create_task(
        title="Написать тесты",
        priority=6,
        tags=["testing", "качество"]
    )
    
    print(f"Создана задача: {{task1.title}} (ID: {{task1.id}})")
    print(f"Создана задача: {{task2.title}} (ID: {{task2.id}})")
    
    # Фильтрация задач
    high_priority = PriorityFilter(7)
    important_tasks = await service.get_filtered_tasks(high_priority)
    
    print(f"\nВажные задачи (приоритет >= 7):")
    for task in important_tasks:
        print(f"- {{task.title}} (приоритет: {{task.priority}})")
    
    # Завершение задачи
    await service.complete_task(task2.id)
    print(f"\nЗадача '{{task2.title}}' завершена!")

if __name__ == "__main__":
    asyncio.run(main())

Особенности: Полная типизация, async/await, SOLID принципы, dependency injection, comprehensive тесты