yaro/inventory
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

CRUDKit update.

Browse source
This commit is contained in:
Yaro Kasear 2025-09-08 11:56:29 -05:00
parent de29d45106
commit f1fa1f2407
8 changed files with 391 additions and 44 deletions
Download patch file Download diff file Expand all files Collapse all files

38
crudkit/core/service.py
View file

@ -18,37 +18,44 @@ class CRUDService(Generic[T]):
def get_query(self):
if self.polymorphic:
poly_model = with_polymorphic(self.model, '*')
return self.session.query(poly_model)
return self.session.query(poly_model), poly_model
else:
base_only = with_polymorphic(self.model, [], flat=True)
return self.session.query(base_only)
return self.session.query(base_only), base_only
def get(self, id: int, include_deleted: bool = False) -> T | None:
obj = self.get_query().filter_by(id=id).first()
if obj is None:
return None
if self.supports_soft_delete and not include_deleted and obj.is_deleted:
return None
return obj
query, root_alias = self.get_query()
if self.supports_soft_delete and not include_deleted:
query = query.filter(getattr(root_alias, "is_deleted") == False)
query = query.filter(getattr(root_alias, "id") == id)
obj = query.first()
return obj or None
def list(self, params=None) -> list[T]:
query = self.get_query()
query, root_alias = self.get_query()
if params:
if self.supports_soft_delete:
include_deleted = False
include_deleted = _is_truthy(params.get('include_deleted'))
if not include_deleted:
query = query.filter(self.model.is_deleted == False)
spec = CRUDSpec(self.model, params)
query = query.filter(getattr(root_alias, "is_deleted") == False)
spec = CRUDSpec(self.model, params, root_alias)
filters = spec.parse_filters()
order_by = spec.parse_sort()
limit, offset = spec.parse_pagination()
for parent, relationship_attr, alias in spec.get_join_paths():
query = query.join(alias, relationship_attr.of_type(alias), isouter=True)
for parent_alias, relationship_attr, target_alias in spec.get_join_paths():
query = query.join(
target_alias,
relationship_attr.of_type(target_alias),
isouter=True
)
for eager in spec.get_eager_loads():
for eager in spec.get_eager_loads(root_alias):
query = query.options(eager)
if filters:
@ -56,6 +63,7 @@ class CRUDService(Generic[T]):
if order_by:
query = query.order_by(*order_by)
query = query.offset(offset).limit(limit)
return query.all()
def create(self, data: dict, actor=None) -> T: