46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
from typing import Type, TypeVar, Generic
|
|
from sqlalchemy.orm import Session
|
|
from crudkit.core.spec import CRUDSpec
|
|
|
|
T = TypeVar("T")
|
|
|
|
class CRUDService(Generic[T]):
|
|
def __init__(self, model: Type[T], session: Session):
|
|
self.model = model
|
|
self.session = session
|
|
|
|
def get(self, id: int) -> T:
|
|
return self.session.get(self.model, id)
|
|
|
|
def list(self, params=None) -> list[T]:
|
|
query = self.session.query(self.model)
|
|
if params:
|
|
spec = CRUDSpec(self.model, params)
|
|
filters = spec.parse_filters()
|
|
order_by = spec.parse_sort()
|
|
limit, offset = spec.parse_pagination()
|
|
|
|
if filters:
|
|
query = query.filter(*filters)
|
|
if order_by:
|
|
query = query.order_by(*order_by)
|
|
query = query.offset(offset).limit(limit)
|
|
return query.all()
|
|
|
|
def create(self, data: dict) -> T:
|
|
obj = self.model(**data)
|
|
self.session.add(obj)
|
|
self.session.commit()
|
|
return obj
|
|
|
|
def update(self, id: int, data: dict) -> T:
|
|
obj = self.get(id)
|
|
for k, v in data.items():
|
|
setattr(obj, k, v)
|
|
self.session.commit()
|
|
return obj
|
|
|
|
def delete(self, id: int):
|
|
obj = self.get(id)
|
|
self.session.delete(obj)
|
|
self.session.commit()
|