65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
"""Base repository with generic CRUD operations"""
|
|
|
|
from typing import TypeVar, Generic, Type, List, Optional, Any
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import select
|
|
from app.db.database import Base as SQLAlchemyBase
|
|
|
|
T = TypeVar("T", bound=SQLAlchemyBase)
|
|
|
|
|
|
class BaseRepository(Generic[T]):
|
|
"""Generic repository for CRUD operations"""
|
|
|
|
def __init__(self, session: Session, model: Type[T]):
|
|
self.session = session
|
|
self.model = model
|
|
|
|
def create(self, **kwargs) -> T:
|
|
"""Create and return new instance"""
|
|
instance = self.model(**kwargs)
|
|
self.session.add(instance)
|
|
self.session.commit()
|
|
self.session.refresh(instance)
|
|
return instance
|
|
|
|
def get_by_id(self, id: Any) -> Optional[T]:
|
|
"""Get instance by primary key"""
|
|
return self.session.query(self.model).filter(self.model.id == id).first()
|
|
|
|
def get_all(self, skip: int = 0, limit: int = 100) -> List[T]:
|
|
"""Get all instances with pagination"""
|
|
return (
|
|
self.session.query(self.model)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def update(self, id: Any, **kwargs) -> Optional[T]:
|
|
"""Update instance by id"""
|
|
instance = self.get_by_id(id)
|
|
if instance:
|
|
for key, value in kwargs.items():
|
|
setattr(instance, key, value)
|
|
self.session.commit()
|
|
self.session.refresh(instance)
|
|
return instance
|
|
|
|
def delete(self, id: Any) -> bool:
|
|
"""Delete instance by id"""
|
|
instance = self.get_by_id(id)
|
|
if instance:
|
|
self.session.delete(instance)
|
|
self.session.commit()
|
|
return True
|
|
return False
|
|
|
|
def exists(self, **kwargs) -> bool:
|
|
"""Check if instance exists with given filters"""
|
|
return self.session.query(self.model).filter_by(**kwargs).first() is not None
|
|
|
|
def count(self, **kwargs) -> int:
|
|
"""Count instances with given filters"""
|
|
return self.session.query(self.model).filter_by(**kwargs).count()
|