Fix paging.

This commit is contained in:
Yaro Kasear 2025-10-07 10:28:11 -05:00
parent 9ca52a6f52
commit 53cc90a74b

View file

@ -205,22 +205,42 @@ class CRUDService(Generic[T]):
# Make sure joins/filters match the real query # Make sure joins/filters match the real query
query = self._apply_firsthop_strategies(query, root_alias, plan) query = self._apply_firsthop_strategies(query, root_alias, plan)
if plan.filters: if plan.filters:
query = query.filters(*plan.filters) query = query.filter(*plan.filters)
order_spec = self._extract_order_spec(root_alias, plan.order_by) order_spec = self._extract_order_spec(root_alias, plan.order_by)
query = query.order_by(*self._order_clauses(order_spec, invert=False))
# We only need the order-by columns for the anchor # Inner subquery must be ordered exactly like the real query
anchor_q = self.session.query(*order_spec.cols) inner = query.order_by(*self._order_clauses(order_spec, invert=False))
# IMPORTANT: anchor_q must use the same FROMs/joins as `query`
anchor_q = anchor_q.select_from(query.subquery()) # IMPORTANT: Build subquery that actually exposes the order-by columns
# under predictable names, then select FROM that and reference subq.c[...]
subq = inner.with_entities(*order_spec.cols).subquery()
# Map the order columns to the subquery columns by key/name
cols_on_subq = []
for col in order_spec.cols:
key = getattr(col, "key", None) or getattr(col, "name", None)
if not key:
# Fallback, but frankly your order cols should have names
raise ValueError("Order-by column is missing a key/name")
cols_on_subq.append(getattr(subq.c, key))
# Now the outer anchor query orders and offsets on the subquery columns
anchor_q = (
self.session
.query(*cols_on_subq)
.select_from(subq)
.order_by(*[
(c.desc() if is_desc else c.asc())
for c, is_desc in zip(cols_on_subq, order_spec.desc)
])
)
offset = max(0, (page - 1) * per_page - 1) offset = max(0, (page - 1) * per_page - 1)
row = anchor_q.offset(offset).limit(1).first() row = anchor_q.offset(offset).limit(1).first()
if not row: if not row:
return None return None
# Row might be a tuple-like; turn into list for _key_predicate return list(row) # tuple-like -> list for _key_predicate
return list(row)
def _apply_not_deleted(self, query, root_alias, params): def _apply_not_deleted(self, query, root_alias, params):
if self.supports_soft_delete and not _is_truthy((params or {}).get("include_deleted")): if self.supports_soft_delete and not _is_truthy((params or {}).get("include_deleted")):