Repository Pattern en FastAPI

Concepto de Repository Pattern

El Repository Pattern es un patrón de diseño arquitectural que actúa como una capa de abstracción entre la lógica de negocio de tu aplicación y la capa de acceso a datos. En el contexto de FastAPI con SQLAlchemy, este patrón encapsula las operaciones de base de datos y proporciona una interfaz más limpia y mantenible para interactuar con los datos.

¿Qué problema resuelve el Repository Pattern?

En aplicaciones FastAPI típicas, es común ver operaciones SQLAlchemy mezcladas directamente en los endpoints o en funciones de utilidad. Esto genera varios problemas:

# Código problemático sin Repository Pattern
@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    # Lógica de base de datos mezclada con el endpoint
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    return user

Este enfoque presenta limitaciones significativas:

• Acoplamiento fuerte entre endpoints y operaciones de base de datos
• Dificultad para realizar testing al no poder simular fácilmente las operaciones
• Repetición de código en múltiples endpoints que realizan consultas similares
• Violación del principio de responsabilidad única al mezclar lógica de API con acceso a datos

Principios fundamentales del Repository Pattern

El Repository Pattern se basa en tres principios clave que lo convierten en una solución elegante:

Encapsulación de acceso a datos: Todas las operaciones relacionadas con una entidad específica se concentran en una única clase repository. Esto significa que si necesitas cambiar cómo se almacenan los usuarios, solo modificas la clase UserRepository.

Abstracción de la persistencia: Los endpoints de tu API no necesitan saber si los datos provienen de SQLAlchemy, MongoDB, o incluso un archivo JSON. El repository proporciona una interfaz consistente independientemente del mecanismo de persistencia subyacente.

Separación de responsabilidades: Cada componente tiene un propósito específico y bien definido:

• Endpoints: Manejan requests HTTP y responses
• Repository: Gestiona operaciones de persistencia
• Modelos: Representan la estructura de datos

Ventajas en aplicaciones FastAPI

El Repository Pattern aporta beneficios específicos al ecosistema FastAPI que mejoran significativamente la calidad del código:

• Testing simplificado: Puedes crear mock repositories fácilmente para tus tests unitarios, eliminando la dependencia de una base de datos real durante las pruebas.

• Flexibilidad de implementación: Cambiar de SQLAlchemy a otro ORM o sistema de persistencia requiere únicamente modificar la implementación del repository, manteniendo intacta la lógica de los endpoints.

• Reutilización de código: Las consultas complejas se escriben una vez en el repository y se reutilizan en múltiples endpoints según sea necesario.

• Mantenibilidad mejorada: Al concentrar toda la lógica de acceso a datos de una entidad en un solo lugar, las modificaciones y correcciones son más directas y predecibles.

Cuándo implementar Repository Pattern

No todas las aplicaciones FastAPI requieren este patrón. Considera implementarlo cuando:

Tu aplicación tiene múltiples endpoints que realizan operaciones similares sobre las mismas entidades. Si tienes varios endpoints que consultan usuarios de diferentes maneras, un UserRepository centralizará estas operaciones.

Necesitas testing robusto de la lógica de negocio sin depender de la base de datos. Los repositories facilitan la creación de dobles de testing que simulan el comportamiento de la persistencia.

Planeas cambiar de tecnología de persistencia en el futuro o necesitas compatibilidad con múltiples sistemas de almacenamiento simultáneamente.

La complejidad de las consultas aumenta y comenzar a tener lógica de base de datos repetida o compleja distribuida por toda la aplicación.

Estructura conceptual básica

Un repository típico en FastAPI sigue una estructura predecible que facilita su comprensión y uso:

# Estructura conceptual - no implementación completa
class UserRepository:
    def __init__(self, db: Session):
        self.db = db
    
    def get_by_id(self, user_id: int) -> Optional[User]:
        # Lógica para obtener usuario por ID
        pass
    
    def get_all(self, skip: int = 0, limit: int = 100) -> List[User]:
        # Lógica para obtener lista paginada
        pass
    
    def create(self, user_data: UserCreate) -> User:
        # Lógica para crear nuevo usuario
        pass
    
    def update(self, user_id: int, user_data: UserUpdate) -> Optional[User]:
        # Lógica para actualizar usuario existente
        pass
    
    def delete(self, user_id: int) -> bool:
        # Lógica para eliminar usuario
        pass

Esta estructura proporciona una interfaz clara y consistente que cualquier desarrollador del equipo puede entender y utilizar de manera efectiva.

El Repository Pattern actúa como el puente arquitectural entre tu lógica de aplicación y la persistencia de datos, manteniendo tu código FastAPI limpio, testeable y mantenible a medida que tu aplicación crece en complejidad.

Implementación y abstracción BD

La implementación práctica del Repository Pattern en FastAPI requiere una estructura bien definida que integre seamlessly con SQLAlchemy y el sistema de inyección de dependencias. Esta sección te guiará a través de una implementación completa que abstrae efectivamente las operaciones de base de datos.

Implementación base del Repository

Comenzamos definiendo una clase repository que encapsula todas las operaciones de persistencia para una entidad específica. La estructura básica sigue un patrón consistente:

from sqlalchemy.orm import Session
from typing import List, Optional
from models import User
from schemas import UserCreate, UserUpdate

class UserRepository:
    def __init__(self, db: Session):
        self.db = db
        self.model = User
    
    def get_by_id(self, user_id: int) -> Optional[User]:
        """Obtiene un usuario por su ID"""
        return self.db.query(self.model).filter(self.model.id == user_id).first()
    
    def get_by_email(self, email: str) -> Optional[User]:
        """Obtiene un usuario por su email"""
        return self.db.query(self.model).filter(self.model.email == email).first()
    
    def get_all(self, skip: int = 0, limit: int = 100) -> List[User]:
        """Obtiene una lista paginada de usuarios"""
        return self.db.query(self.model).offset(skip).limit(limit).all()
    
    def create(self, user_data: UserCreate) -> User:
        """Crea un nuevo usuario"""
        db_user = User(**user_data.model_dump())
        self.db.add(db_user)
        self.db.commit()
        self.db.refresh(db_user)
        return db_user
    
    def update(self, user_id: int, user_data: UserUpdate) -> Optional[User]:
        """Actualiza un usuario existente"""
        db_user = self.get_by_id(user_id)
        if db_user:
            update_data = user_data.model_dump(exclude_unset=True)
            for field, value in update_data.items():
                setattr(db_user, field, value)
            self.db.commit()
            self.db.refresh(db_user)
        return db_user
    
    def delete(self, user_id: int) -> bool:
        """Elimina un usuario"""
        db_user = self.get_by_id(user_id)
        if db_user:
            self.db.delete(db_user)
            self.db.commit()
            return True
        return False

Abstracción mediante interfaces

Para maximizar la flexibilidad y facilitar el testing, podemos definir una interfaz abstracta que establezca el contrato que debe cumplir cualquier implementación de repository:

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

T = TypeVar('T')

class BaseRepository(ABC, Generic[T]):
    @abstractmethod
    def get_by_id(self, entity_id: int) -> Optional[T]:
        pass
    
    @abstractmethod
    def get_all(self, skip: int = 0, limit: int = 100) -> List[T]:
        pass
    
    @abstractmethod
    def create(self, entity_data: dict) -> T:
        pass
    
    @abstractmethod
    def update(self, entity_id: int, entity_data: dict) -> Optional[T]:
        pass
    
    @abstractmethod
    def delete(self, entity_id: int) -> bool:
        pass

class SQLAlchemyUserRepository(BaseRepository[User]):
    def __init__(self, db: Session):
        self.db = db
        self.model = User
    
    def get_by_id(self, user_id: int) -> Optional[User]:
        return self.db.query(self.model).filter(self.model.id == user_id).first()
    
    # ... implementación de otros métodos

Esta abstracción permite crear diferentes implementaciones del mismo repository, como una versión para testing que use datos en memoria o una implementación alternativa para otro sistema de persistencia.

Integración con Dependency Injection

FastAPI utiliza un sistema de inyección de dependencias que se integra perfectamente con el Repository Pattern. Creamos una función que actúa como factory del repository:

from database import get_db

def get_user_repository(db: Session = Depends(get_db)) -> UserRepository:
    """Factory function para crear instancia del UserRepository"""
    return UserRepository(db)

# Uso en endpoints
@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(
    user_id: int,
    user_repo: UserRepository = Depends(get_user_repository)
):
    """Obtiene un usuario por ID usando el repository"""
    user = user_repo.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    return user

@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(
    user_data: UserCreate,
    user_repo: UserRepository = Depends(get_user_repository)
):
    """Crea un nuevo usuario"""
    # Verificar si el email ya existe
    existing_user = user_repo.get_by_email(user_data.email)
    if existing_user:
        raise HTTPException(status_code=400, detail="El email ya está registrado")
    
    return user_repo.create(user_data)

Manejo de transacciones y errores

Un repository bien implementado debe manejar adecuadamente las transacciones y los errores que puedan surgir durante las operaciones de base de datos:

from sqlalchemy.exc import SQLAlchemyError
from fastapi import HTTPException

class UserRepository:
    def __init__(self, db: Session):
        self.db = db
        self.model = User
    
    def create_with_transaction(self, user_data: UserCreate) -> User:
        """Crea usuario con manejo explícito de transacciones"""
        try:
            db_user = User(**user_data.model_dump())
            self.db.add(db_user)
            self.db.commit()
            self.db.refresh(db_user)
            return db_user
        except SQLAlchemyError as e:
            self.db.rollback()
            raise HTTPException(
                status_code=500, 
                detail=f"Error al crear usuario: {str(e)}"
            )
    
    def bulk_create(self, users_data: List[UserCreate]) -> List[User]:
        """Crea múltiples usuarios en una sola transacción"""
        try:
            db_users = [User(**user_data.model_dump()) for user_data in users_data]
            self.db.add_all(db_users)
            self.db.commit()
            
            for db_user in db_users:
                self.db.refresh(db_user)
            
            return db_users
        except SQLAlchemyError as e:
            self.db.rollback()
            raise HTTPException(
                status_code=500,
                detail=f"Error en creación masiva: {str(e)}"
            )

Consultas complejas y filtros

Los repositories son el lugar ideal para encapsular consultas complejas que involucren múltiples condiciones, joins, o agregaciones:

class UserRepository:
    # ... métodos anteriores
    
    def find_by_criteria(
        self, 
        name: Optional[str] = None,
        email_domain: Optional[str] = None,
        is_active: Optional[bool] = None,
        skip: int = 0,
        limit: int = 100
    ) -> List[User]:
        """Busca usuarios aplicando múltiples filtros opcionales"""
        query = self.db.query(self.model)
        
        if name:
            query = query.filter(self.model.name.ilike(f"%{name}%"))
        
        if email_domain:
            query = query.filter(self.model.email.ilike(f"%@{email_domain}"))
        
        if is_active is not None:
            query = query.filter(self.model.is_active == is_active)
        
        return query.offset(skip).limit(limit).all()
    
    def count_by_status(self) -> dict:
        """Cuenta usuarios agrupados por estado activo/inactivo"""
        active_count = self.db.query(self.model).filter(self.model.is_active == True).count()
        inactive_count = self.db.query(self.model).filter(self.model.is_active == False).count()
        
        return {
            "active": active_count,
            "inactive": inactive_count,
            "total": active_count + inactive_count
        }

Repository genérico reutilizable

Para evitar duplicación de código entre diferentes repositories, podemos crear una clase base genérica que implemente operaciones comunes:

from typing import Type, TypeVar, Generic
from sqlalchemy.orm import Session
from sqlalchemy.ext.declarative import DeclarativeMeta

ModelType = TypeVar("ModelType", bound=DeclarativeMeta)
CreateSchemaType = TypeVar("CreateSchemaType")
UpdateSchemaType = TypeVar("UpdateSchemaType")

class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    def __init__(self, model: Type[ModelType], db: Session):
        self.model = model
        self.db = db
    
    def get_by_id(self, entity_id: int) -> Optional[ModelType]:
        return self.db.query(self.model).filter(self.model.id == entity_id).first()
    
    def get_all(self, skip: int = 0, limit: int = 100) -> List[ModelType]:
        return self.db.query(self.model).offset(skip).limit(limit).all()
    
    def create(self, entity_data: CreateSchemaType) -> ModelType:
        db_entity = self.model(**entity_data.model_dump())
        self.db.add(db_entity)
        self.db.commit()
        self.db.refresh(db_entity)
        return db_entity
    
    def delete(self, entity_id: int) -> bool:
        db_entity = self.get_by_id(entity_id)
        if db_entity:
            self.db.delete(db_entity)
            self.db.commit()
            return True
        return False

# Implementación específica que hereda funcionalidad común
class UserRepository(BaseRepository[User, UserCreate, UserUpdate]):
    def __init__(self, db: Session):
        super().__init__(User, db)
    
    def get_by_email(self, email: str) -> Optional[User]:
        """Método específico para usuarios"""
        return self.db.query(self.model).filter(self.model.email == email).first()

Integración completa en endpoints

La implementación final integra todos estos conceptos en endpoints limpios y mantenibles:

@app.get("/users", response_model=List[UserResponse])
def list_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
    name: Optional[str] = Query(None),
    is_active: Optional[bool] = Query(None),
    user_repo: UserRepository = Depends(get_user_repository)
):
    """Lista usuarios con filtros opcionales"""
    if name or is_active is not None:
        return user_repo.find_by_criteria(
            name=name, 
            is_active=is_active, 
            skip=skip, 
            limit=limit
        )
    return user_repo.get_all(skip=skip, limit=limit)

@app.put("/users/{user_id}", response_model=UserResponse)
def update_user(
    user_id: int,
    user_data: UserUpdate,
    user_repo: UserRepository = Depends(get_user_repository)
):
    """Actualiza un usuario existente"""
    updated_user = user_repo.update(user_id, user_data)
    if not updated_user:
        raise HTTPException(status_code=404, detail="Usuario no encontrado")
    return updated_user

Esta implementación del Repository Pattern proporciona una abstracción sólida que separa completamente la lógica de persistencia de los endpoints, facilitando el mantenimiento, testing y evolución de tu aplicación FastAPI.