Какую проблему решает использование паттерна Repository над ORM?

Какую проблему решает паттерн Repository над ORM

Паттерн Repository решает несколько критических проблем, которые возникают при прямом использовании ORM в бизнес-логике приложения.

Основная проблема: Прямая зависимость от ORM

Без Repository ваш код выглядит так:

# Плохо: прямое использование ORM в бизнес-логике
from sqlalchemy import Session
from app.models import User

def create_user(session: Session, email: str, name: str):
    user = User(email=email, name=name)  # Привязка к SQLAlchemy
    session.add(user)
    session.commit()
    return user

def get_active_users(session: Session):
    return session.query(User).filter(User.is_active == True).all()  # SQLAlchemy везде

def update_user(session: Session, user_id: int, data: dict):
    user = session.query(User).get(user_id)  # Привязка к конкретной ORM
    if user:
        for key, value in data.items():
            setattr(user, key, value)
        session.commit()
    return user

Проблемы этого подхода:

1. Высокая связанность с конкретной ORM — если переходите с SQLAlchemy на Tortoise или Django ORM, нужно переписать весь код
2. Сложно тестировать — нельзя использовать простые mock-объекты, нужно мокировать весь session
3. Бизнес-логика смешана с доступом к БД — нарушение принципа Single Responsibility
4. Дублирование запросов — сложные queries повторяются в разных частях кода
5. Сложность изменения логики работы с БД — изменения в БД влияют на всю бизнес-логику

Решение: Паттерн Repository

Repository создает абстракцию над ORM:

# Хорошо: Repository паттерн
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass

@dataclass
class User:
    """Модель домена, НЕ привязана к ORM"""
    id: int
    email: str
    name: str
    is_active: bool

class UserRepository(ABC):
    """Интерфейс репозитория - не зависит от конкретной БД"""
    
    @abstractmethod
    def create(self, email: str, name: str) -> User:
        pass
    
    @abstractmethod
    def get_by_id(self, user_id: int) -> Optional[User]:
        pass
    
    @abstractmethod
    def get_active_users(self) -> List[User]:
        pass
    
    @abstractmethod
    def update(self, user_id: int, data: dict) -> Optional[User]:
        pass
    
    @abstractmethod
    def delete(self, user_id: int) -> bool:
        pass

# Реализация для SQLAlchemy
from sqlalchemy.orm import Session
from app.models import UserModel  # ORM модель

class SQLAlchemyUserRepository(UserRepository):
    def __init__(self, session: Session):
        self.session = session
    
    def create(self, email: str, name: str) -> User:
        db_user = UserModel(email=email, name=name)
        self.session.add(db_user)
        self.session.commit()
        return User(
            id=db_user.id,
            email=db_user.email,
            name=db_user.name,
            is_active=db_user.is_active
        )
    
    def get_by_id(self, user_id: int) -> Optional[User]:
        db_user = self.session.query(UserModel).get(user_id)
        if not db_user:
            return None
        return User(
            id=db_user.id,
            email=db_user.email,
            name=db_user.name,
            is_active=db_user.is_active
        )
    
    def get_active_users(self) -> List[User]:
        db_users = self.session.query(UserModel).filter(
            UserModel.is_active == True
        ).all()
        return [
            User(
                id=db_user.id,
                email=db_user.email,
                name=db_user.name,
                is_active=db_user.is_active
            )
            for db_user in db_users
        ]
    
    def update(self, user_id: int, data: dict) -> Optional[User]:
        db_user = self.session.query(UserModel).get(user_id)
        if not db_user:
            return None
        for key, value in data.items():
            if hasattr(db_user, key):
                setattr(db_user, key, value)
        self.session.commit()
        return User(
            id=db_user.id,
            email=db_user.email,
            name=db_user.name,
            is_active=db_user.is_active
        )
    
    def delete(self, user_id: int) -> bool:
        db_user = self.session.query(UserModel).get(user_id)
        if not db_user:
            return False
        self.session.delete(db_user)
        self.session.commit()
        return True

# Альтернативная реализация для PostgreSQL с простым SQL
import asyncpg

class PostgreSQLUserRepository(UserRepository):
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool
    
    async def create(self, email: str, name: str) -> User:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                'INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id, email, name, is_active',
                email, name
            )
            return User(
                id=row['id'],
                email=row['email'],
                name=row['name'],
                is_active=row['is_active']
            )
    
    async def get_by_id(self, user_id: int) -> Optional[User]:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                'SELECT id, email, name, is_active FROM users WHERE id = $1',
                user_id
            )
            if not row:
                return None
            return User(
                id=row['id'],
                email=row['email'],
                name=row['name'],
                is_active=row['is_active']
            )
    
    # ... остальные методы

Бизнес-логика, независимая от ORM

# UseCase - не зависит от конкретной реализации Repository
class CreateUserUseCase:
    def __init__(self, user_repo: UserRepository):
        self.user_repo = user_repo  # Зависит от интерфейса, не от реализации
    
    def execute(self, email: str, name: str) -> User:
        # Проверка валидации
        if not email or not name:
            raise ValueError("Email and name are required")
        
        # Проверка уникальности
        existing = self.user_repo.get_by_email(email)
        if existing:
            raise ValueError("User with this email already exists")
        
        # Создание пользователя
        return self.user_repo.create(email, name)

class GetActiveUsersUseCase:
    def __init__(self, user_repo: UserRepository):
        self.user_repo = user_repo
    
    def execute(self) -> List[User]:
        users = self.user_repo.get_active_users()
        return sorted(users, key=lambda u: u.name)

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

# Тестовая реализация Repository
class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self.users = {}
        self.next_id = 1
    
    def create(self, email: str, name: str) -> User:
        user = User(
            id=self.next_id,
            email=email,
            name=name,
            is_active=True
        )
        self.users[self.next_id] = user
        self.next_id += 1
        return user
    
    def get_by_id(self, user_id: int) -> Optional[User]:
        return self.users.get(user_id)
    
    # ... остальные методы

# Тестирование
import pytest

def test_create_user_use_case():
    # Используем в памяти репозиторий для тестов
    repo = InMemoryUserRepository()
    use_case = CreateUserUseCase(repo)
    
    user = use_case.execute(email="test@example.com", name="John")
    
    assert user.email == "test@example.com"
    assert user.name == "John"
    assert user.id == 1

def test_create_user_duplicate_email():
    repo = InMemoryUserRepository()
    use_case = CreateUserUseCase(repo)
    
    use_case.execute(email="test@example.com", name="John")
    
    with pytest.raises(ValueError):
        use_case.execute(email="test@example.com", name="Jane")

Преимущества Repository паттерна

1. Слабая связанность - можно менять ORM без изменения бизнес-логики
2. Простота тестирования - легко создать тестовую реализацию
3. Переиспользование логики - сложные queries хранятся в одном месте
4. Четкое разделение - Database Layer и Business Logic отделены
5. Гибкость - легко переключаться между разными реализациями БД
6. Maintainability - изменения в БД изолированы от бизнес-логики

Dependency Injection

# FastAPI пример
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

def get_user_repository(session: Session = Depends(get_db)) -> UserRepository:
    return SQLAlchemyUserRepository(session)

@app.post("/users")
def create_user(
    email: str,
    name: str,
    user_repo: UserRepository = Depends(get_user_repository)
):
    use_case = CreateUserUseCase(user_repo)
    return use_case.execute(email, name)

Резюме

Repository паттерн решает проблему привязки бизнес-логики к конкретной ORM через создание абстракции (интерфейса), которая позволяет:

• Заменять реализацию БД без изменения кода
• Легко тестировать через подставленные реализации
• Сосредоточиться на бизнес-логике вместо деталей доступа к БД
• Переиспользовать сложные query в одном месте