Что такое паттерн Репозиторий (Repository)?

Паттерн Репозиторий (Repository Pattern)

Repository Pattern — это архитектурный паттерн, который абстрагирует работу с хранилищем данных (базой данных, файлами, кэшем). Репозиторий действует как посредник между бизнес-логикой (domain layer) и источником данных (data layer), предоставляя методы для CRUD операций (Create, Read, Update, Delete).

Основная идея

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

• Более тестируемым — легко мокировать БД
• Более читаемым — вся логика работы с данными в одном месте
• Более гибким — легко менять источник данных

Простой пример

# ❌ БЕЗ Repository Pattern — SQL везде
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()
engine = create_engine(sqlite:///db.sqlite)
Session = sessionmaker(bind=engine)

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

# Логика разбросана по коду
session = Session()
user = session.query(User).filter(User.id == 1).first()
user.name = "Новое имя"
session.commit()

# ✓ С Repository Pattern — всё в репозитории
class UserRepository:
    """Репозиторий для работы с пользователями"""
    
    def __init__(self, session):
        self.session = session
    
    def get_by_id(self, user_id: int) -> User:
        """Получить пользователя по ID"""
        return self.session.query(User).filter(User.id == user_id).first()
    
    def get_by_email(self, email: str) -> User:
        """Получить пользователя по email"""
        return self.session.query(User).filter(User.email == email).first()
    
    def get_all(self) -> list[User]:
        """Получить всех пользователей"""
        return self.session.query(User).all()
    
    def create(self, name: str, email: str) -> User:
        """Создать новго пользователя"""
        user = User(name=name, email=email)
        self.session.add(user)
        self.session.commit()
        return user
    
    def update(self, user_id: int, name: str, email: str) -> User:
        """Обновить пользователя"""
        user = self.get_by_id(user_id)
        if user:
            user.name = name
            user.email = email
            self.session.commit()
        return user
    
    def delete(self, user_id: int) -> bool:
        """Удалить пользователя"""
        user = self.get_by_id(user_id)
        if user:
            self.session.delete(user)
            self.session.commit()
            return True
        return False

# Теперь используем репозиторий
repo = UserRepository(session)
user = repo.get_by_id(1)
repo.update(1, "Иван Петров", "ivan@example.com")

Пример с интерфейсом (Interface/ABC)

from abc import ABC, abstractmethod
from typing import Generic, TypeVar, List

T = TypeVar(T)  # Generics для типизации

class IRepository(ABC, Generic[T]):
    """Интерфейс репозитория"""
    
    @abstractmethod
    def get_by_id(self, id: int) -> T | None:
        pass
    
    @abstractmethod
    def get_all(self) -> List[T]:
        pass
    
    @abstractmethod
    def create(self, entity: T) -> T:
        pass
    
    @abstractmethod
    def update(self, id: int, entity: T) -> T | None:
        pass
    
    @abstractmethod
    def delete(self, id: int) -> bool:
        pass

class UserRepository(IRepository[User]):
    """Конкретная реализация для пользователей"""
    
    def __init__(self, session):
        self.session = session
    
    def get_by_id(self, id: int) -> User | None:
        return self.session.query(User).filter(User.id == id).first()
    
    def get_all(self) -> List[User]:
        return self.session.query(User).all()
    
    def create(self, entity: User) -> User:
        self.session.add(entity)
        self.session.commit()
        return entity
    
    def update(self, id: int, entity: User) -> User | None:
        user = self.get_by_id(id)
        if user:
            for key, value in vars(entity).items():
                if not key.startswith(_):
                    setattr(user, key, value)
            self.session.commit()
        return user
    
    def delete(self, id: int) -> bool:
        user = self.get_by_id(id)
        if user:
            self.session.delete(user)
            self.session.commit()
            return True
        return False

class ProductRepository(IRepository[Product]):
    """Другой репозиторий, для продуктов"""
    # Аналогичная реализация для Product
    pass

Использование в бизнес-логике

class UserService:
    """Сервис для работы с пользователями (бизнес-логика)"""
    
    def __init__(self, user_repo: IRepository[User]):
        self.repo = user_repo
    
    def register_user(self, name: str, email: str) -> User:
        """Регистрация нового пользователя"""
        # Проверяем, не существует ли уже пользователь
        existing = self.repo.get_all()  # Или специальный метод
        if any(u.email == email for u in existing):
            raise ValueError("Пользователь с таким email уже существует")
        
        # Создаём пользователя
        user = User(name=name, email=email)
        return self.repo.create(user)
    
    def deactivate_user(self, user_id: int) -> bool:
        """Деактивировать пользователя"""
        return self.repo.delete(user_id)

# Использование
service = UserService(user_repo)
service.register_user("Иван", "ivan@example.com")

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

from unittest.mock import Mock

def test_register_user():
    # Мокируем репозиторий
    mock_repo = Mock(spec=IRepository[User])
    mock_repo.get_all.return_value = []
    mock_repo.create.return_value = User(id=1, name="Иван", email="ivan@example.com")
    
    # Тестируем сервис с мокированным репозиторием
    service = UserService(mock_repo)
    user = service.register_user("Иван", "ivan@example.com")
    
    assert user.id == 1
    assert user.name == "Иван"
    mock_repo.create.assert_called_once()

Преимущества и недостатки

Преимущества:

• Отделение логики от БД
• Простое тестирование через моки
• Единая точка изменения запросов
• Переиспользование логики
• Легко менять источник данных

Недостатки:

• Дополнительный слой абстракции (немного боilerplate кода)
• Может быть избыточным для простых проектов
• ORM может уже быть абстракцией

Repository Pattern — это фундаментальный паттерн чистой архитектуры, который делает код более модульным, тестируемым и поддерживаемым.