Fixed two of the most annoying bugs in the universe: Extra sessions and malfunctioning params.

crudkit/core/service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
+from flask import current_app
 from typing import Any, Callable, Type, TypeVar, Generic, Optional, Protocol, runtime_checkable, cast
 from sqlalchemy import and_, func, inspect, or_, text
 from sqlalchemy.engine import Engine, Connection
@@ -133,24 +134,25 @@ class CRUDService(Generic[T]):
         self.polymorphic = polymorphic
         self.supports_soft_delete = hasattr(model, 'is_deleted')
 
-        # Derive engine WITHOUT leaking a session/connection
-        bind = getattr(session_factory, "bind", None)
-        if bind is None:
-            tmp_sess = session_factory()
-            try:
-                bind = tmp_sess.get_bind()
-            finally:
-                try:
-                    tmp_sess.close()
-                except Exception:
-                    pass
-
-        eng: Engine = bind.engine if isinstance(bind, Connection) else cast(Engine, bind)
-        self.backend = backend or make_backend_info(eng)
+        self._backend: Optional[BackendInfo] = backend
 
     @property
     def session(self) -> Session:
-        return self._session_factory()
+        """Always return the Flask-scoped Session if available; otherwise the provided factory."""
+        try:
+            sess = current_app.extensions["crudkit"]["Session"]
+            return sess
+        except Exception:
+            return self._session_factory()
+
+    @property
+    def backend(self) -> BackendInfo:
+        """Resolve backend info lazily against the active session's engine."""
+        if self._backend is None:
+            bind = self.session.get_bind()
+            eng: Engine = bind.engine if isinstance(bind, Connection) else cast(Engine, bind)
+            self._backend = make_backend_info(eng)
+        return self._backend
 
     def get_query(self):
         if self.polymorphic:
@@ -237,85 +239,46 @@ class CRUDService(Generic[T]):
         include_total: bool = True,
     ) -> "SeekWindow[T]":
         """
-        Transport-agnostic keyset pagination that preserves all the goodies from `list()`:
-          - filters, includes, joins, field projection, eager loading, soft-delete
-          - deterministic ordering (user sort + PK tiebreakers)
-          - forward/backward seek via `key` and `backward`
-        Returns a SeekWindow with items, first/last keys, order spec, limit, and optional total.
+        Keyset pagination with relationship-safe filtering/sorting.
+        Always JOIN all CRUDSpec-discovered paths first; then apply filters, sort, seek.
         """
-        self._debug_bind("seek_window")
         session = self.session
         query, root_alias = self.get_query()
 
-        # Normalize requested fields and compile projection (may skip later to avoid conflicts)
+        # Requested fields → projection + optional loaders
         fields = _normalize_fields_param(params)
         expanded_fields, proj_opts = compile_projection(self.model, fields) if fields else ([], [])
 
         spec = CRUDSpec(self.model, params or {}, root_alias)
 
+        # Parse all inputs so join_paths are populated
         filters = spec.parse_filters()
         order_by = spec.parse_sort()
-
-        # Field parsing for root load_only fallback
         root_fields, rel_field_names, root_field_names = spec.parse_fields()
+        spec.parse_includes()
+        join_paths = tuple(spec.get_join_paths())
 
-        # Soft delete filter
+        # Soft delete
         query = self._apply_not_deleted(query, root_alias, params)
 
-        # Apply filters first
-        if filters:
-            query = query.filter(*filters)
-
-        # Includes + join paths (dotted fields etc.)
-        spec.parse_includes()
-        join_paths = tuple(spec.get_join_paths())  # iterable of (path, relationship_attr, target_alias)
-
-        # Relationship names required by ORDER BY / WHERE
-        sql_hops: set[str] = _paths_needed_for_sql(order_by, filters, join_paths)
-        # Also include relationships mentioned directly in the sort spec
-        sql_hops |= _hops_from_sort(params)
-
-        # First-hop relationship names implied by dotted projection fields
-        proj_hops: set[str] = _paths_from_fields(fields)
-
-        # Root column projection
+        # Root column projection (load_only)
         only_cols = [c for c in root_fields if isinstance(c, InstrumentedAttribute)]
         if only_cols:
             query = query.options(Load(root_alias).load_only(*only_cols))
 
-        # Relationship handling per path (avoid loader strategy conflicts)
+        # JOIN all resolved paths, hydrate from the join
         used_contains_eager = False
-        joined_names: set[str] = set()
-
-        for _path, relationship_attr, target_alias in join_paths:
-            rel_attr = cast(InstrumentedAttribute, relationship_attr)
-            name = relationship_attr.key
-            if name in sql_hops:
-                # Needed for WHERE/ORDER BY: join + hydrate from that join
-                query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
-                query = query.options(contains_eager(rel_attr, alias=target_alias))
-                used_contains_eager = True
-                joined_names.add(name)
-            elif name in proj_hops:
-                # Display-only: bulk-load efficiently, no join
-                query = query.options(selectinload(rel_attr))
-                joined_names.add(name)
-
-        # Force-join any SQL-needed relationships that weren't in join_paths
-        missing_sql = sql_hops - joined_names
-        for name in missing_sql:
-            rel_attr = cast(InstrumentedAttribute, getattr(root_alias, name))
-            query = query.join(rel_attr, isouter=True)
-            query = query.options(contains_eager(rel_attr))
+        for _base_alias, rel_attr, target_alias in join_paths:
+            query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+            query = query.options(contains_eager(rel_attr, alias=target_alias))
             used_contains_eager = True
-            joined_names.add(name)
 
-        # Apply projection loader options only if they won't conflict with contains_eager
-        if proj_opts and not used_contains_eager:
-            query = query.options(*proj_opts)
+        # Filters AFTER joins → no cartesian products
+        if filters:
+            query = query.filter(*filters)
 
-        # Order + limit
-        order_spec = self._extract_order_spec(root_alias, order_by)  # SA 2.x helper
+        # Order spec (with PK tie-breakers for stability)
+        order_spec = self._extract_order_spec(root_alias, order_by)
         limit, _ = spec.parse_pagination()
         if limit is None:
             effective_limit = 50
@@ -324,13 +287,13 @@ class CRUDService(Generic[T]):
         else:
             effective_limit = limit
 
-        # Keyset predicate
+        # Seek predicate from cursor key (if any)
         if key:
             pred = self._key_predicate(order_spec, key, backward)
             if pred is not None:
                 query = query.filter(pred)
 
-        # Apply ordering. For backward, invert SQL order then reverse in-memory for display.
+        # Apply ORDER and LIMIT. Backward is SQL-inverted + reverse in-memory.
         if not backward:
             clauses = [(c.desc() if is_desc else c.asc()) for c, is_desc in zip(order_spec.cols, order_spec.desc)]
             query = query.order_by(*clauses)
@@ -344,9 +307,9 @@ class CRUDService(Generic[T]):
                 query = query.limit(effective_limit)
             items = list(reversed(query.all()))
 
-        # Tag projection so your renderer knows what fields were requested
+        # Projection meta tag for renderers
         if fields:
-            proj = list(dict.fromkeys(fields))  # dedupe, preserve order
+            proj = list(dict.fromkeys(fields))
             if "id" not in proj and hasattr(self.model, "id"):
                 proj.insert(0, "id")
         else:
@@ -369,12 +332,9 @@ class CRUDService(Generic[T]):
                 except Exception:
                     pass
 
-        # Boundary keys for cursor encoding in the API layer
-        # When ORDER BY includes related columns (e.g., owner.first_name),
-        # pluck values from the related object we hydrated with contains_eager/selectinload.
+        # Cursor key pluck: support related columns we hydrated via contains_eager
         def _pluck_key_from_obj(obj: Any) -> list[Any]:
             vals: list[Any] = []
-            # Build a quick map: selectable -> relationship name
             alias_to_rel: dict[Any, str] = {}
             for _p, relationship_attr, target_alias in join_paths:
                 sel = getattr(target_alias, "selectable", None)
@@ -382,20 +342,17 @@ class CRUDService(Generic[T]):
                     alias_to_rel[sel] = relationship_attr.key
 
             for col in order_spec.cols:
-                key = getattr(col, "key", None) or getattr(col, "name", None)
-                # Try root attribute first
-                if key and hasattr(obj, key):
-                    vals.append(getattr(obj, key))
+                keyname = getattr(col, "key", None) or getattr(col, "name", None)
+                if keyname and hasattr(obj, keyname):
+                    vals.append(getattr(obj, keyname))
                     continue
-                # Try relationship hop by matching the column's table/selectable
                 table = getattr(col, "table", None)
                 relname = alias_to_rel.get(table)
-                if relname and key:
+                if relname and keyname:
                     relobj = getattr(obj, relname, None)
-                    if relobj is not None and hasattr(relobj, key):
-                        vals.append(getattr(relobj, key))
+                    if relobj is not None and hasattr(relobj, keyname):
+                        vals.append(getattr(relobj, keyname))
                         continue
-                # Give up: unsupported expression for cursor purposes
                 raise ValueError("unpluckable")
             return vals
 
@@ -403,33 +360,24 @@ class CRUDService(Generic[T]):
             first_key = _pluck_key_from_obj(items[0]) if items else None
             last_key  = _pluck_key_from_obj(items[-1]) if items else None
         except Exception:
-            # If we can't derive cursor keys (e.g., ORDER BY expression/aggregate),
-            # disable cursors for this response rather than exploding.
             first_key = None
             last_key = None
 
-        # Optional total that’s safe under JOINs (COUNT DISTINCT ids)
+        # Count DISTINCT ids with mirrored joins
         total = None
         if include_total:
             base = session.query(getattr(root_alias, "id"))
             base = self._apply_not_deleted(base, root_alias, params)
+            # same joins as above for correctness
+            for _base_alias, rel_attr, target_alias in join_paths:
+                base = base.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
             if filters:
                 base = base.filter(*filters)
-            # Mirror join structure for any SQL-needed relationships
-            for _path, relationship_attr, target_alias in join_paths:
-                if relationship_attr.key in sql_hops:
-                    rel_attr = cast(InstrumentedAttribute, relationship_attr)
-                    base = base.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
-            # Also mirror any forced joins
-            for name in (sql_hops - {ra.key for _p, ra, _a in join_paths}):
-                rel_attr = cast(InstrumentedAttribute, getattr(root_alias, name))
-                base = base.join(rel_attr, isouter=True)
-
             total = session.query(func.count()).select_from(
                 base.order_by(None).distinct().subquery()
             ).scalar() or 0
 
-        window_limit_for_body = 0 if effective_limit is None and limit == 0 else (effective_limit or 50)
+        window_limit_for_body = 0 if effective_limit is None and (limit == 0) else (effective_limit or 50)
 
         if log.isEnabledFor(logging.DEBUG):
             log.debug("QUERY: %s", str(query))
@@ -474,65 +422,53 @@ class CRUDService(Generic[T]):
         return [*order_by, *pk_cols]
 
     def get(self, id: int, params=None) -> T | None:
-        """Fetch a single row by id with conflict-free eager loading and clean projection."""
-        self._debug_bind("get")
+        """
+        Fetch a single row by id with conflict-free eager loading and clean projection.
+        Always JOIN any paths that CRUDSpec resolved for filters/fields/includes so
+        related-column filters never create cartesian products.
+        """
         query, root_alias = self.get_query()
 
         # Defaults so we can build a projection even if params is None
+        req_fields: list[str] = _normalize_fields_param(params)
         root_fields: list[Any] = []
         root_field_names: dict[str, str] = {}
         rel_field_names: dict[tuple[str, ...], list[str]] = {}
-        req_fields: list[str] = _normalize_fields_param(params)
 
-        # Soft-delete guard
+        # Soft-delete guard first
         query = self._apply_not_deleted(query, root_alias, params)
 
         spec = CRUDSpec(self.model, params or {}, root_alias)
 
-        # Optional extra filters (in addition to id); keep parity with list()
+        # Parse everything so CRUDSpec records any join paths it needed to resolve
         filters = spec.parse_filters()
-        if filters:
-            query = query.filter(*filters)
-
-        # Always filter by id
-        query = query.filter(getattr(root_alias, "id") == id)
-
-        # Includes + join paths we may need
+        # no ORDER BY for get()
+        if params:
+            root_fields, rel_field_names, root_field_names = spec.parse_fields()
         spec.parse_includes()
         join_paths = tuple(spec.get_join_paths())
 
-        # Field parsing to enable root load_only
-        if params:
-            root_fields, rel_field_names, root_field_names = spec.parse_fields()
-
-        # Decide which relationship paths are needed for SQL vs display-only
-        # For get(), there is no ORDER BY; only filters might force SQL use.
-        sql_hops = _paths_needed_for_sql(order_by=None, filters=filters, join_paths=join_paths)
-        proj_hops = _paths_from_fields(req_fields)
-
-        # Root column projection
+        # Root-column projection (load_only)
         only_cols = [c for c in root_fields if isinstance(c, InstrumentedAttribute)]
         if only_cols:
             query = query.options(Load(root_alias).load_only(*only_cols))
 
-        # Relationship handling per path: avoid loader strategy conflicts
+        # JOIN all discovered paths up front; hydrate via contains_eager
         used_contains_eager = False
-        for _path, relationship_attr, target_alias in join_paths:
-            rel_attr = cast(InstrumentedAttribute, relationship_attr)
-            name = relationship_attr.key
-            if name in sql_hops:
-                # Needed in WHERE: join + hydrate from the join
-                query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
-                query = query.options(contains_eager(rel_attr, alias=target_alias))
-                used_contains_eager = True
-            elif name in proj_hops:
-                # Display-only: bulk-load efficiently
-                query = query.options(selectinload(rel_attr))
-            else:
-                pass
+        for _base_alias, rel_attr, target_alias in join_paths:
+            query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+            query = query.options(contains_eager(rel_attr, alias=target_alias))
+            used_contains_eager = True
+
+        # Apply filters (joins are in place → no cartesian products)
+        if filters:
+            query = query.filter(*filters)
+
+        # And the id filter
+        query = query.filter(getattr(root_alias, "id") == id)
 
         # Projection loader options compiled from requested fields.
-        # Skip if we used contains_eager to avoid strategy conflicts.
+        # Skip if we used contains_eager to avoid loader-strategy conflicts.
         expanded_fields, proj_opts = compile_projection(self.model, req_fields) if req_fields else ([], [])
         if proj_opts and not used_contains_eager:
             query = query.options(*proj_opts)
@@ -541,7 +477,7 @@ class CRUDService(Generic[T]):
 
         # Emit exactly what the client requested (plus id), or a reasonable fallback
         if req_fields:
-            proj = list(dict.fromkeys(req_fields))  # dedupe, preserve order
+            proj = list(dict.fromkeys(req_fields))
             if "id" not in proj and hasattr(self.model, "id"):
                 proj.insert(0, "id")
         else:
@@ -569,17 +505,20 @@ class CRUDService(Generic[T]):
         return obj or None
 
     def list(self, params=None) -> list[T]:
-        """Offset/limit listing with smart relationship loading and clean projection."""
-        self._debug_bind("list")
+        """
+        Offset/limit listing with relationship-safe filtering.
+        We always JOIN every CRUDSpec-discovered path before applying filters/sorts.
+        """
         query, root_alias = self.get_query()
 
         # Defaults so we can reference them later even if params is None
+        req_fields: list[str] = _normalize_fields_param(params)
         root_fields: list[Any] = []
         root_field_names: dict[str, str] = {}
         rel_field_names: dict[tuple[str, ...], list[str]] = {}
-        req_fields: list[str] = _normalize_fields_param(params)
 
         if params:
+            # Soft delete
             query = self._apply_not_deleted(query, root_alias, params)
 
             spec = CRUDSpec(self.model, params or {}, root_alias)
@@ -587,84 +526,56 @@ class CRUDService(Generic[T]):
             order_by = spec.parse_sort()
             limit, offset = spec.parse_pagination()
 
-            # Includes + join paths we might need
+            # Includes / fields (populates join_paths)
+            root_fields, rel_field_names, root_field_names = spec.parse_fields()
             spec.parse_includes()
             join_paths = tuple(spec.get_join_paths())
 
-            # Field parsing for load_only on root columns
-            root_fields, rel_field_names, root_field_names = spec.parse_fields()
-
-            if filters:
-                query = query.filter(*filters)
-
-            # Determine which relationship paths are needed for SQL vs display-only
-            sql_hops = _paths_needed_for_sql(order_by, filters, join_paths)
-            sql_hops |= _hops_from_sort(params)  # ensure sort-driven joins exist
-            proj_hops = _paths_from_fields(req_fields)
-
-            # Root column projection
+            # Root column projection (load_only)
             only_cols = [c for c in root_fields if isinstance(c, InstrumentedAttribute)]
             if only_cols:
                 query = query.options(Load(root_alias).load_only(*only_cols))
 
-            # Relationship handling per path
+            # JOIN all paths we resolved and hydrate them from the join
             used_contains_eager = False
-            joined_names: set[str] = set()
-
-            for _path, relationship_attr, target_alias in join_paths:
-                rel_attr = cast(InstrumentedAttribute, relationship_attr)
-                name = relationship_attr.key
-                if name in sql_hops:
-                    # Needed for WHERE/ORDER BY: join + hydrate from the join
-                    query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
-                    query = query.options(contains_eager(rel_attr, alias=target_alias))
-                    used_contains_eager = True
-                    joined_names.add(name)
-                elif name in proj_hops:
-                    # Display-only: no join, bulk-load efficiently
-                    query = query.options(selectinload(rel_attr))
-                    joined_names.add(name)
-
-            # Force-join any SQL-needed relationships that weren't in join_paths
-            missing_sql = sql_hops - joined_names
-            for name in missing_sql:
-                rel_attr = cast(InstrumentedAttribute, getattr(root_alias, name))
-                query = query.join(rel_attr, isouter=True)
-                query = query.options(contains_eager(rel_attr))
+            for _base_alias, rel_attr, target_alias in join_paths:
+                query = query.join(target_alias, rel_attr.of_type(target_alias), isouter=True)
+                query = query.options(contains_eager(rel_attr, alias=target_alias))
                 used_contains_eager = True
-                joined_names.add(name)
 
-            # MSSQL requires ORDER BY when OFFSET is used (SQLA uses OFFSET for limit/offset)
+            # Filters AFTER joins → no cartesian products
+            if filters:
+                query = query.filter(*filters)
+
+            # MSSQL requires ORDER BY when OFFSET is used; ensure stable PK tie-breakers
             paginating = (limit is not None) or (offset is not None and offset != 0)
             if paginating and not order_by and self.backend.requires_order_by_for_offset:
                 order_by = self._default_order_by(root_alias)
-
             if order_by:
-                query = query.order_by(*order_by)
+                query = query.order_by(*self._stable_order_by(root_alias, order_by))
 
-            # Only apply offset/limit when not None and not zero
+            # Offset/limit
             if offset is not None and offset != 0:
                 query = query.offset(offset)
             if limit is not None and limit > 0:
                 query = query.limit(limit)
 
-            # Projection loader options compiled from requested fields.
-            # Skip if we used contains_eager to avoid loader-strategy conflicts.
+            # Projection loaders only if we didn’t use contains_eager
             expanded_fields, proj_opts = compile_projection(self.model, req_fields) if req_fields else ([], [])
             if proj_opts and not used_contains_eager:
                 query = query.options(*proj_opts)
 
         else:
-            # No params means no filters/sorts/limits; still honor projection loaders if any
+            # No params; still honor projection loaders if any
             expanded_fields, proj_opts = compile_projection(self.model, req_fields) if req_fields else ([], [])
             if proj_opts:
                 query = query.options(*proj_opts)
 
         rows = query.all()
 
-        # Emit exactly what the client requested (plus id), or a reasonable fallback
+        # Build projection meta for renderers
         if req_fields:
-            proj = list(dict.fromkeys(req_fields))  # dedupe while preserving order
+            proj = list(dict.fromkeys(req_fields))
             if "id" not in proj and hasattr(self.model, "id"):
                 proj.insert(0, "id")
         else:
@@ -692,7 +603,6 @@ class CRUDService(Generic[T]):
 
         return rows
 
-
     def create(self, data: dict, actor=None) -> T:
         session = self.session
         obj = self.model(**data)

crudkit/integrations/flask.py

@@ -26,32 +26,6 @@ def init_app(app: Flask, *, runtime: CRUDKitRuntime | None = None, config: type[
     try:
         bound_engine = getattr(SessionFactory, "bind", None) or getattr(SessionFactory, "kw", {}).get("bind") or engine
         pool = bound_engine.pool
-
-        from sqlalchemy import event
-
-        @event.listens_for(pool, "checkout")
-        def _on_checkout(dbapi_conn, conn_record, conn_proxy):
-            sz = pool.size()
-            chk = pool.checkedout()
-            try:
-                conns_in_pool = pool.checkedin()
-            except Exception:
-                conns_in_pool = "?"
-            print(f"POOL CHECKOUT: Pool size: {sz}  Connections in pool: {conns_in_pool} "
-                  f"Current Overflow: {pool.overflow()} Current Checked out connections: {chk} "
-                  f"engine id= {id(bound_engine)}")
-
-        @event.listens_for(pool, "checkin")
-        def _on_checkin(dbapi_conn, conn_record):
-            sz = pool.size()
-            chk = pool.checkedout()
-            try:
-                conns_in_pool = pool.checkedin()
-            except Exception:
-                conns_in_pool = "?"
-            print(f"POOL CHECKIN: Pool size: {sz}  Connections in pool: {conns_in_pool} "
-                  f"Current Overflow: {pool.overflow()} Current Checked out connections: {chk} "
-                  f"engine id= {id(bound_engine)}")
     except Exception as e:
         print(f"[crudkit.init_app] Failed to attach pool listeners: {e}")
 

inventory/__init__.py

@@ -15,25 +15,6 @@ from .routes.index import init_index_routes
 from .routes.listing import init_listing_routes
 from .routes.entry import init_entry_routes
 
-def _bind_pool_debug(engine):
-    pool = engine.pool
-    eid = id(engine)
-
-    @event.listens_for(pool, "checkout")
-    def _on_checkout(dbapi_con, con_record, con_proxy):
-        try:
-            print(f"POOL CHECKOUT: {pool.status()} engine id= {eid}")
-        except Exception:
-            # pool.status is safe on SQLA 2.x, but let's be defensive
-            print(f"POOL CHECKOUT (no status) engine id= {eid}")
-
-    @event.listens_for(pool, "checkin")
-    def _on_checkin(dbapi_con, con_record):
-        try:
-            print(f"POOL CHECKIN: {pool.status()} engine id= {eid}")
-        except Exception:
-            print(f"POOL CHECKIN (no status) engine id= {eid}")
-
 def create_app(config_cls=crudkit.DevConfig) -> Flask:
     app = Flask(__name__)
 
@@ -43,8 +24,6 @@ def create_app(config_cls=crudkit.DevConfig) -> Flask:
     from sqlalchemy import event
 
     engine = runtime.engine
-    print(f"CRUDKit engine id={id(runtime.engine)} url={runtime.engine.url}")
-    _bind_pool_debug(runtime.engine)   # ← attach to the real engine’s pool
 
     # quick status endpoint you can hit while clicking around
     @app.get("/_db_status")

inventory/routes/index.py

@@ -22,8 +22,7 @@ def init_index_routes(app):
             "fields": [
                 "start_time",
                 "contact.label",
-                "work_item.label",
-                "work_item.device_type.description"
+                "work_item.label"
             ],
             "sort": "start_time"
         })
@@ -65,8 +64,7 @@ def init_index_routes(app):
             {"field": "contact.label", "label": "Contact",
              "link": {"endpoint": "entry.entry", "params": {"id": "{contact.id}", "model": "user"}}},
             {"field": "work_item.label", "label": "Work Item",
-             "link": {"endpoint": "entry.entry", "params": {"id": "{work_item.id}", "model": "inventory"}}},
-            {"field": "work_item.device_type.description", "label": "Device Type"}
+             "link": {"endpoint": "entry.entry", "params": {"id": "{work_item.id}", "model": "inventory"}}}
         ]
 
         logs = render_table(work_logs, columns=columns, opts={"object_class": "worklog"})

inventory/routes/listing.py

@@ -18,10 +18,11 @@ def init_listing_routes(app):
             abort(404)
 
         # read query args
-        limit = int(request.args.get("limit", 15))
-        sort  = request.args.get("sort")  # <- capture sort from URL
+        limit  = request.args.get("limit", None)
+        limit  = int(limit) if (limit is not None and str(limit).isdigit()) else 15
+        sort   = request.args.get("sort")
+        fields_qs = request.args.get("fields")
         cursor = request.args.get("cursor")
-        # your decode returns (key, _desc, backward) in this project
         key, _desc, backward = decode_cursor(cursor)
 
         # base spec per model
@@ -102,48 +103,42 @@ def init_listing_routes(app):
                 {"when": {"field": "complete", "is": False}, "class": "table-danger"}
             ]
 
-        # overlay URL-provided sort if present
+        # Build params to feed CRUDService (flat dict; parse_filters expects flat keys)
+        params = dict(spec)
+
+        # overlay fields from query (?fields=...)
+        if fields_qs:
+            params["fields"] = [p.strip() for p in fields_qs.split(",") if p.strip()]
+
+        # overlay sort from query (?sort=...)
         if sort:
-            spec["sort"] = sort
+            params["sort"] = sort
+
+        # limit semantics: 0 means "unlimited" in your service layer
+        params["limit"] = limit
+
+        # forward *all other* query params as filters (flat), excluding known control keys
+        CONTROL_KEYS = {"limit", "cursor", "sort", "fields"}
+        for k, v in request.args.items():
+            if k in CONTROL_KEYS:
+                continue
+            if v is None or v == "":
+                continue
+            params[k] = v
 
         service = crudkit.crud.get_service(cls)
-        try:
-            rt = app.extensions["crudkit"]["runtime"]
-            rt_engine = rt.engine
-        except Exception:
-            rt_engine = None
 
-        try:
-            SessionFactory = app.extensions["crudkit"].get("SessionFactory")
-            sf_engine = getattr(SessionFactory, "bind", None) or getattr(SessionFactory, "kw", {}).get("bind")
-        except Exception:
-            sf_engine = None
-
-        try:
-            bind = service.session.get_bind()
-            svc_engine = getattr(bind, "engine", bind)
-        except Exception:
-            svc_engine = None
-
-        print(
-            "LISTING ENGINES: "
-            f"runtime={id(rt_engine) if rt_engine else None}  "
-            f"session_factory.bind={id(sf_engine) if sf_engine else None}  "
-            f"service.bind={id(svc_engine) if svc_engine else None}"
-        )
-        # include limit and go
-        window = service.seek_window(spec | {"limit": limit}, key=key, backward=backward, include_total=True)
+        window = service.seek_window(params, key=key, backward=backward, include_total=True)
 
         table = render_table(window.items, columns=columns,
                              opts={"object_class": model, "row_classes": row_classes})
 
-        # pass sort through so templates can preserve it on pager links, if they care
         pagination_ctx = {
             "limit": window.limit,
             "total": window.total,
             "next_cursor": encode_cursor(window.last_key, list(window.order.desc), backward=False),
             "prev_cursor": encode_cursor(window.first_key, list(window.order.desc), backward=True),
-            "sort": sort or spec.get("sort")  # expose current sort to the template
+            "sort": params.get("sort")  # expose current sort to the template
         }
 
         return render_template("listing.html", model=model, table=table, pagination=pagination_ctx)