global-repository 1.0.0

Repositorio Base extencion de alchemy para funciones generales de BD

Global Repository

Repositorio base genérico para SQLAlchemy con operaciones CRUD, filtros avanzados, paginación y búsqueda.

Descripción

global-repository es una librería que proporciona una clase BaseRepository genérica que extiende SQLAlchemy para simplificar las operaciones de base de datos. Está diseñada para ser heredada por repositorios específicos de cada entidad.

Características

• Operaciones CRUD completas: create, read, update, delete
• Filtros avanzados: 12 operadores de comparación
• Búsqueda flexible: por uno o múltiples campos
• Paginación integrada: con metadatos completos
• Ordenamiento: ASC/DESC simple o múltiple
• Conteo y agregación: con filtros opcionales
• Operaciones bulk: actualización y eliminación masiva
• Transacciones: soporte para ejecutar funciones en transacción

---

Instalación

Como dependencia de proyecto

# pyproject.toml
[tool.poetry.dependencies]
global-repository = "^1.0.0"

O con pip:

pip install global-repository

Desarrollo local

# Clonar el repositorio
git clone <repo-url>
cd base-repository

# Instalar dependencias
pip install -e .

# Ejecutar tests
pytest tests/

---

Estructura del Proyecto

src/global_repository/
├── __init__.py          # Exports públicos del paquete
├── enums.py             # OrderDirection, ComparisonOperator
├── dataclasses.py       # FilterCondition, PaginationResult, OrderBy
└── base_repository.py   # Clase BaseRepository principal

---

Uso en tu Proyecto

1. Configurar SQLAlchemy

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

class Base(DeclarativeBase):
    pass

# Conexión a la base de datos
engine = create_engine("sqlite:///mi_db.sqlite")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_session() -> Session:
    return SessionLocal()

2. Definir tus modelos

# models.py
from sqlalchemy import Column, Integer, String, Boolean
from database import Base

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True)
    is_active = Column(Boolean, default=True)

3. Crear repositorios específicos

# repositories.py
from sqlalchemy.orm import Session
from src.global_repository import BaseRepository
from models import User

class UserRepository(BaseRepository[User]):
    def __init__(self, session: Session):
        super().__init__(User, session)
    
    # Puedes agregar métodos específicos de User
    def get_active_users(self):
        return self.where(is_active=True)

4. Usar en tu aplicación

from database import get_session
from repositories import UserRepository

def main():
    session = get_session()
    repo = UserRepository(session)
    
    # Crear
    user = User(name="Juan", email="[e-mail]")
    created = repo.create(user)
    
    # Consultar con filtros
    active_users = repo.where(is_active=True)
    
    # Paginación
    result = repo.get_all_paginated(page=1, page_size=10)
    print(f"Total: {result.total}, Página: {result.page}")
    
    # Búsqueda
    results = repo.search(field="name", value="Juan", exact=False)
    
    session.close()

if __name__ == "__main__":
    main()

---

Referencia de API

Enums

Enum	Descripción
OrderDirection.ASC	Orden ascendente
OrderDirection.DESC	Orden descendente
ComparisonOperator.EQ	Igual a (=)
ComparisonOperator.NE	Diferente de (!=)
ComparisonOperator.GT	Mayor que (>)
ComparisonOperator.GE	Mayor o igual (>=)
ComparisonOperator.LT	Menor que (<)
ComparisonOperator.LE	Menor o igual (<=)
ComparisonOperator.LIKE	Como (LIKE %value%)
ComparisonOperator.ILIKE	Como sin distinción de mayúsculas
ComparisonOperator.IN	En lista
ComparisonOperator.NOT_IN	No en lista
ComparisonOperator.IS_NULL	Es NULL
ComparisonOperator.IS_NOT_NULL	No es NULL

Data Classes

# FilterCondition
FilterCondition(field="status", operator=ComparisonOperator.EQ, value="active")

# OrderBy
OrderBy(field="name", direction=OrderDirection.ASC)

# PaginationResult (retorno de get_all_paginated)
result.items        # Lista de elementos
result.total        # Total sin paginar
result.page         # Página actual
result.page_size    # Tamaño de página
result.total_pages  # Total de páginas
result.has_next     # Hay siguiente página
result.has_previous # Hay página anterior

Métodos del Repositorio

Método	Descripción
create(obj)	Crea un registro
create_many(objects)	Crea múltiples registros
get_by_id(id)	Obtiene por ID
get_all()	Obtiene todos
update(obj)	Actualiza un registro
delete(obj)	Elimina un registro
delete_by_id(id)	Elimina por ID
exists(id)	Verifica existencia
filter(conditions, ...)	Filtra con condiciones
filter_one(conditions)	Un resultado con filtros
search(field, value, ...)	Búsqueda en un campo
search_multiple_fields(...)	Búsqueda en múltiples campos
get_all_paginated(...)	Resultados paginados
get_all_ordered(order_by)	Resultados ordenados
count(conditions)	Conteo con filtros
count_all()	Total de registros
get_first(...)	Primer registro
get_last(order_by)	Último registro
bulk_update(objects)	Actualización masiva
bulk_delete(objects)	Eliminación masiva
where(**kwargs)	Filtro rápido por igualdad
where_one(**kwargs)	Un resultado por igualdad
where_not(**kwargs)	Exclusión por igualdad
where_in(field, values)	Filtrar por lista
where_not_in(field, values)	Excluir por lista
where_null(field)	Filtrar nulos
where_not_null(field)	Filtrar no nulos

---

Múltiples Bases de Datos

Sí, es compatible. El diseño es stateless respecto a la conexión — cada instancia de repositorio recibe la session en su constructor, por lo que puedes trabajar con múltiples bases de datos.

Ejemplo: Múltiples conexiones

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Base de datos principal
engine_main = create_engine("postgresql://user:pass@localhost/main_db")
SessionMain = sessionmaker(bind=engine_main)

# Base de datos de reportes
engine_reports = create_engine("postgresql://user:pass@localhost/reports_db")
SessionReports = sessionmaker(bind=engine_reports)

# Base de datos legacy
engine_legacy = create_engine("sqlite:///legacy.db")
SessionLegacy = sessionmaker(bind=engine_legacy)

# repositories.py
from src.global_repository import BaseRepository

class UserRepository(BaseRepository[User]):
    def __init__(self, session):
        super().__init__(User, session)

class ReportRepository(BaseRepository[Report]):
    def __init__(self, session):
        super().__init__(Report, session)

class LegacyCustomerRepository(BaseRepository[LegacyCustomer]):
    def __init__(self, session):
        super().__init__(LegacyCustomer, session)

# uso.py
def get_user_repos():
    session_main = SessionMain()
    return UserRepository(session_main)

def get_report_repos():
    session_reports = SessionReports()
    return ReportRepository(session_reports)

def get_legacy_repos():
    session_legacy = SessionLegacy()
    return LegacyCustomerRepository(session_legacy)

# Uso
users = get_user_repos().get_all()
reports = get_report_repos().where(status="pending")

Patrón recomendado: Unit of Work

class UnitOfWork:
    def __init__(self, session_factory):
        self.session = session_factory()
        self.users = UserRepository(self.session)
        self.reports = ReportRepository(self.session)
    
    def commit(self):
        self.session.commit()
    
    def rollback(self):
        self.session.rollback()
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.session.close()

# Uso
with UnitOfWork(SessionMain) as uow:
    uow.users.create(User(name="Nuevo"))
    uow.reports.create(Report(title="Reporte 1"))
    uow.commit()

---

Requisitos

• Python >= 3.13
• SQLAlchemy >= 2.0.0

Licencia

Copyright (c) 2026 Erick Damian Gonzalez Aranda - NeuronexoTec