97 lines
2.7 KiB
Python
97 lines
2.7 KiB
Python
"""
|
|
Base repository class with common CRUD operations.
|
|
"""
|
|
|
|
from typing import TypeVar, Generic, Type, Optional, List, Any, Dict
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func, asc, desc
|
|
from uuid import UUID
|
|
|
|
from ..models import Base
|
|
|
|
T = TypeVar("T", bound=Base)
|
|
|
|
|
|
class BaseRepository(Generic[T]):
|
|
"""
|
|
Base repository providing common CRUD operations.
|
|
|
|
Subclasses should set `model` class attribute to the SQLAlchemy model.
|
|
"""
|
|
|
|
model: Type[T]
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def get_by_id(self, id: Any) -> Optional[T]:
|
|
"""Get entity by primary key."""
|
|
return self.db.query(self.model).filter(self.model.id == id).first()
|
|
|
|
def get_all(
|
|
self,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
order_by: str = None,
|
|
order_desc: bool = False,
|
|
) -> List[T]:
|
|
"""Get all entities with pagination and optional ordering."""
|
|
query = self.db.query(self.model)
|
|
|
|
if order_by and hasattr(self.model, order_by):
|
|
column = getattr(self.model, order_by)
|
|
query = query.order_by(desc(column) if order_desc else asc(column))
|
|
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
def count(self) -> int:
|
|
"""Count total entities."""
|
|
return self.db.query(func.count(self.model.id)).scalar() or 0
|
|
|
|
def create(self, **kwargs) -> T:
|
|
"""Create a new entity."""
|
|
entity = self.model(**kwargs)
|
|
self.db.add(entity)
|
|
self.db.flush() # Flush to get ID without committing
|
|
return entity
|
|
|
|
def update(self, entity: T, **kwargs) -> T:
|
|
"""Update an existing entity."""
|
|
for key, value in kwargs.items():
|
|
if hasattr(entity, key):
|
|
setattr(entity, key, value)
|
|
self.db.flush()
|
|
return entity
|
|
|
|
def delete(self, entity: T) -> None:
|
|
"""Delete an entity."""
|
|
self.db.delete(entity)
|
|
self.db.flush()
|
|
|
|
def delete_by_id(self, id: Any) -> bool:
|
|
"""Delete entity by ID. Returns True if deleted, False if not found."""
|
|
entity = self.get_by_id(id)
|
|
if entity:
|
|
self.delete(entity)
|
|
return True
|
|
return False
|
|
|
|
def exists(self, id: Any) -> bool:
|
|
"""Check if entity exists by ID."""
|
|
return self.db.query(
|
|
self.db.query(self.model).filter(self.model.id == id).exists()
|
|
).scalar()
|
|
|
|
def commit(self) -> None:
|
|
"""Commit the current transaction."""
|
|
self.db.commit()
|
|
|
|
def rollback(self) -> None:
|
|
"""Rollback the current transaction."""
|
|
self.db.rollback()
|
|
|
|
def refresh(self, entity: T) -> T:
|
|
"""Refresh entity from database."""
|
|
self.db.refresh(entity)
|
|
return entity
|