115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any, Generic, TypeVar
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.database import Base
|
|
|
|
ModelType = TypeVar("ModelType", bound=Base)
|
|
|
|
|
|
class BaseRepository(Generic[ModelType]):
|
|
"""Base repository with common CRUD operations."""
|
|
|
|
def __init__(self, db: Session, model: type[ModelType]) -> None:
|
|
self.db = db
|
|
self.model = model
|
|
|
|
def get_by_id(self, entity_id: str | uuid.UUID) -> ModelType | None:
|
|
"""Get an entity by its primary key."""
|
|
if isinstance(entity_id, str):
|
|
entity_id = uuid.UUID(entity_id)
|
|
return self.db.get(self.model, entity_id)
|
|
|
|
def get_all(
|
|
self,
|
|
offset: int = 0,
|
|
limit: int = 100,
|
|
filters: dict[str, Any] | None = None,
|
|
order_by: str | None = None,
|
|
order_desc: bool = False,
|
|
) -> list[ModelType]:
|
|
"""Get all entities with optional filtering, pagination, and ordering."""
|
|
query = select(self.model)
|
|
|
|
if filters:
|
|
for key, value in filters.items():
|
|
if hasattr(self.model, key) and value is not None:
|
|
query = query.where(getattr(self.model, key) == value)
|
|
|
|
if order_by and hasattr(self.model, order_by):
|
|
col = getattr(self.model, order_by)
|
|
query = query.order_by(col.desc() if order_desc else col.asc())
|
|
|
|
query = query.offset(offset).limit(limit)
|
|
result = self.db.execute(query)
|
|
return list(result.scalars().all())
|
|
|
|
def count(self, filters: dict[str, Any] | None = None) -> int:
|
|
"""Count entities with optional filtering."""
|
|
query = select(func.count()).select_from(self.model)
|
|
|
|
if filters:
|
|
for key, value in filters.items():
|
|
if hasattr(self.model, key) and value is not None:
|
|
query = query.where(getattr(self.model, key) == value)
|
|
|
|
result = self.db.execute(query)
|
|
return result.scalar_one()
|
|
|
|
def create(self, entity: ModelType) -> ModelType:
|
|
"""Create a new entity."""
|
|
self.db.add(entity)
|
|
self.db.flush()
|
|
self.db.refresh(entity)
|
|
return entity
|
|
|
|
def create_many(self, entities: list[ModelType]) -> list[ModelType]:
|
|
"""Create multiple entities."""
|
|
self.db.add_all(entities)
|
|
self.db.flush()
|
|
for entity in entities:
|
|
self.db.refresh(entity)
|
|
return entities
|
|
|
|
def update(self, entity: ModelType, update_data: dict[str, Any]) -> ModelType:
|
|
"""Update an entity with given data."""
|
|
for key, value in update_data.items():
|
|
if hasattr(entity, key) and value is not None:
|
|
setattr(entity, key, value)
|
|
self.db.flush()
|
|
self.db.refresh(entity)
|
|
return entity
|
|
|
|
def delete(self, entity: ModelType) -> None:
|
|
"""Delete an entity."""
|
|
self.db.delete(entity)
|
|
self.db.flush()
|
|
|
|
def delete_by_id(self, entity_id: str | uuid.UUID) -> bool:
|
|
"""Delete an entity by its ID. Returns True if deleted."""
|
|
entity = self.get_by_id(entity_id)
|
|
if entity:
|
|
self.delete(entity)
|
|
return True
|
|
return False
|
|
|
|
def exists(self, entity_id: str | uuid.UUID) -> bool:
|
|
"""Check if an entity exists by ID."""
|
|
if isinstance(entity_id, str):
|
|
entity_id = uuid.UUID(entity_id)
|
|
query = select(func.count()).select_from(self.model).where(self.model.id == entity_id)
|
|
result = self.db.execute(query)
|
|
return result.scalar_one() > 0
|
|
|
|
def commit(self) -> None:
|
|
"""Commit the current transaction."""
|
|
self.db.commit()
|
|
|
|
def rollback(self) -> None:
|
|
"""Rollback the current transaction."""
|
|
self.db.rollback()
|