feat: add row filters engine

Author: lc-spxl

policies/celine/dataset.rego

@@ -193,6 +193,13 @@ allow if {
     input.action.name in ["query", "read"]
 }
 
+allow if {
+    is_internal
+    is_user
+    "viewers" in input.subject.groups
+    input.action.name in ["query", "read"]
+}
+
 reason := "internal dataset - manager group granted" if {
     allow
     is_internal

src/celine/dataset/api/dataset_query/executor.py

@@ -3,8 +3,11 @@
 
 import json
 import logging
+
+import httpx
+import sqlglot
 from typing import Optional, Dict, Sequence, List
-from fastapi import HTTPException
+from fastapi import HTTPException, Request
 from sqlalchemy import RowMapping, Table, text, select, func
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.exc import DBAPIError
@@ -19,11 +22,12 @@
 )
 from celine.dataset.security.models import AuthenticatedUser
 from celine.dataset.api.dataset_query.parser import parse_sql_query
-from celine.dataset.api.dataset_query.user_filter import (
-    get_user_filter_column,
-    inject_user_filter,
-    is_admin_user,
+from celine.dataset.api.dataset_query.row_filters import (
+    apply_row_filter_plans,
+    get_row_filter_registry,
+    get_row_filter_specs,
 )
+from celine.dataset.api.dataset_query.row_filters.utils import is_admin_user
 
 logger = logging.getLogger(__name__)
 
@@ -103,20 +107,15 @@ async def execute_query(
     - SQL validated (SELECT-only, table allowlist)
     - LIMIT/OFFSET enforced server-side
     - hard row cap applied
-    - user filtering applied (if userFilterColumn defined)
+    - row-level filters applied (pluggable governance handlers)
     """
-    # ------------------------------------------------------------------
-    # Validate SQL
-    # ------------------------------------------------------------------
-
     if raw_sql is None or raw_sql.strip() == "":
         raise HTTPException(400, "sql query not provided")
 
     logger.debug(f"Parsing raw SQL: {raw_sql}")
     try:
         parsed = parse_sql_query(raw_sql)
-    except HTTPException as exc:
-        logger.error(f"SQL validation failed: {exc}")
+    except HTTPException:
         raise
     except Exception as exc:
         logger.exception("SQL validation failed")
@@ -126,12 +125,16 @@ async def execute_query(
         raise HTTPException(400, "Query references no datasets")
 
     datasets = await resolve_datasets_for_tables(db=db, table_names=parsed.tables)
+
     tables_map: dict[str, str] = {}
-    user_filters: List[dict] = []  # Collect filters to apply
+    row_filter_plans = []
+
+    registry = get_row_filter_registry()
 
     for ref_table, ds in datasets.items():
         if not ds.expose:
             raise HTTPException(403, "Dataset not available")
+
         await enforce_dataset_access(entry=ds, user=user)
 
         if ds.backend_config is None:
@@ -148,45 +151,80 @@ async def execute_query(
         logger.debug(f"Mapped SQL table {ref_table} -> {phy_table_name}")
         tables_map[ref_table] = phy_table_name
 
-        # ------------------------------------------------------------------
-        # Check for user filtering requirement
-        # ------------------------------------------------------------------
-        filter_column = get_user_filter_column(ds)
-        if filter_column:
-            if user is None:
+        specs = get_row_filter_specs(ds)
+        if not specs:
+            continue
+
+        if user is None:
+            raise HTTPException(
+                401,
+                f"Dataset {ref_table} requires authentication for row filtering",
+            )
+
+        if is_admin_user(user):
+            continue
+
+        for spec in specs:
+            handler_name = spec.get("handler")
+            args = spec.get("args") or {}
+            if not isinstance(handler_name, str) or not handler_name:
                 raise HTTPException(
-                    401,
-                    f"Dataset {ref_table} requires authentication for user filtering",
+                    500,
+                    f"Invalid row filter spec for dataset {ref_table}: missing handler",
+                )
+            if not isinstance(args, dict):
+                raise HTTPException(
+                    500,
+                    f"Invalid row filter spec for dataset {ref_table}: args must be object",
                 )
 
-            # Admins bypass user filtering
-            if not is_admin_user(user):
-                user_filters.append(
-                    {
-                        "table": phy_table_name,
-                        "column": filter_column,
-                        "user_sub": user.sub,
-                    }
+            try:
+                plan = await registry.resolve_with_cache(
+                    handler_name=handler_name,
+                    table=phy_table_name,
+                    user=user,
+                    args=args,
+                    request_context={},
+                )
+            except KeyError:
+                logger.error(
+                    f"Unknown row filter handler '{handler_name}' for dataset {ref_table}"
+                )
+                raise HTTPException(
+                    500,
+                    f"Unknown row filter handler '{handler_name}' for dataset {ref_table}",
                 )
-                logger.debug(
-                    f"User filter required for {ref_table}: "
-                    f"{filter_column} = {user.sub}"
+            except httpx.HTTPError:
+                logger.error(f"Row filter resolution failed for dataset {ref_table}")
+                raise HTTPException(
+                    403,
+                    f"Row filter resolution failed for dataset {ref_table}",
+                )
+            except Exception as e:
+                logger.error(f"Row filter handler failed: {e}")
+                raise HTTPException(
+                    500,
+                    f"Row filter handler '{handler_name}' failed for dataset {ref_table}",
                 )
 
-    # Replace tables ID with physical tables
-    complete_sql = parsed.to_sql(tables_map=tables_map)
-    logger.debug(f"Complete SQL (before user filter): {complete_sql}")
+            row_filter_plans.append(plan)
 
-    # ------------------------------------------------------------------
-    # Inject user filters
-    # ------------------------------------------------------------------
-    if user_filters:
-        complete_sql = inject_user_filter(complete_sql, user_filters)
-        logger.debug(f"Complete SQL (after user filter): {complete_sql}")
+    # Logical -> physical substitution
+    complete_sql = parsed.to_sql(tables_map=tables_map)
+    logger.debug(f"Complete SQL (after table mapping): {complete_sql}")
+
+    # Apply row-level filters
+    if row_filter_plans:
+        try:
+            ast = sqlglot.parse_one(complete_sql)
+            ast = apply_row_filter_plans(ast, row_filter_plans)
+            complete_sql = ast.sql()
+        except Exception:
+            logger.exception("Failed to apply row filters")
+            raise HTTPException(500, "Failed to apply row filters") from None
+        logger.debug(f"Complete SQL (after row filters): {complete_sql}")
 
-    # ------------------------------------------------------------------
     # Pagination & caps
-    # ------------------------------------------------------------------
     limit = _clamp_limit(limit)
     offset = max(offset, 0)
 
@@ -204,40 +242,29 @@ async def execute_query(
         ) AS q
     """
 
-    # ------------------------------------------------------------------
     # Execute count
-    # ------------------------------------------------------------------
     try:
-        total = await execute_scalar_with_timeout(
-            db,
-            count_sql,
-        )
-    except HTTPException as e:
-        logger.error(f"Query count failed: {e}")
+        total = await execute_scalar_with_timeout(db, count_sql)
+    except HTTPException:
         raise
-    except Exception as exc:  # safety net
+    except Exception:
         logger.exception("Count query failed")
         raise HTTPException(500, "Query failed") from None
 
-    # ------------------------------------------------------------------
     # Execute data query
-    # ------------------------------------------------------------------
     try:
         rows = await execute_rows_with_timeout(
             db,
             paginated_sql,
             {"limit": limit, "offset": offset},
         )
-    except HTTPException as e:
-        logger.error(f"Query execution failed: {e}")
+    except HTTPException:
         raise
-    except Exception as exc:  # safety net
-        logger.error("Query execution failed: {e}")
+    except Exception:
+        logger.exception("Query execution failed")
         raise HTTPException(500, "Query execution failed") from None
 
-    # ------------------------------------------------------------------
     # Post-process rows (geometry → GeoJSON)
-    # ------------------------------------------------------------------
     items = []
     for r in rows:
         row = dict(r)

src/celine/dataset/api/dataset_query/row_filters/__init__.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from .registry import RowFilterRegistry, get_row_filter_registry
+from .specs import get_row_filter_specs
+from .apply import apply_row_filter_plans
+from .models import RowFilterPlan
+
+__all__ = [
+    "RowFilterRegistry",
+    "get_row_filter_registry",
+    "RowFilterPlan",
+    "get_row_filter_specs",
+    "apply_row_filter_plans",
+]

src/celine/dataset/api/dataset_query/row_filters/apply.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import logging
+from typing import Iterable
+
+import sqlglot
+from sqlglot import exp
+
+from celine.dataset.api.dataset_query.row_filters.models import RowFilterPlan
+
+logger = logging.getLogger(__name__)
+
+
+def _table_name(table: exp.Table) -> str:
+    # table.this is Identifier; may contain dots if set that way
+    ident = table.args.get("this")
+    if isinstance(ident, exp.Identifier):
+        return ident.this
+    return table.sql()
+
+
+def _qualify_columns(expr: exp.Expression, alias: str) -> exp.Expression:
+    """Qualify unqualified Column nodes in expr with alias."""
+    e = expr.copy()
+    for col in e.find_all(exp.Column):
+        if col.args.get("table") is None:
+            col.set("table", exp.Identifier(this=alias, quoted=False))
+    return e
+
+
+def _add_where(select: exp.Select, condition: exp.Expression) -> None:
+    existing = select.args.get("where")
+    if isinstance(existing, exp.Where):
+        new_cond = exp.And(this=existing.this, expression=condition)
+        existing.set("this", new_cond)
+    else:
+        select.set("where", exp.Where(this=condition))
+
+
+def _tables_in_select(select: exp.Select) -> list[exp.Table]:
+    tables: list[exp.Table] = []
+    for t in select.find_all(exp.Table):
+        anc = t.find_ancestor(exp.Select)
+        if anc is select:
+            tables.append(t)
+    return tables
+
+
+def apply_row_filter_plans(ast: exp.Expression, plans: Iterable[RowFilterPlan]) -> exp.Expression:
+    """Apply row filter plans to an AST (returns a modified copy)."""
+    plans_by_table: dict[str, list[RowFilterPlan]] = {}
+    for p in plans:
+        plans_by_table.setdefault(p.table, []).append(p)
+
+    out = ast.copy()
+
+    # If any plan is deny -> inject FALSE predicate at top-level
+    for ps in plans_by_table.values():
+        for p in ps:
+            if p.kind == "deny":
+                top = out.find(exp.Select)
+                if top is None:
+                    return out
+                _add_where(top, exp.Boolean(this=False))
+                return out
+
+    for select in out.find_all(exp.Select):
+        tables = _tables_in_select(select)
+        for table in tables:
+            name = _table_name(table)
+            if name not in plans_by_table:
+                continue
+
+            alias = table.alias_or_name
+            for plan in plans_by_table[name]:
+                if plan.kind != "predicate" or plan.predicate_template is None:
+                    continue
+                cond = _qualify_columns(plan.predicate_template, alias)
+                _add_where(select, cond)
+
+    return out

src/celine/dataset/api/dataset_query/row_filters/cache.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from typing import Any, Callable, Generic, Optional, TypeVar
+
+T = TypeVar("T")
+
+
+@dataclass
+class _CacheEntry(Generic[T]):
+    value: T
+    expires_at: float
+
+
+class TTLCache(Generic[T]):
+    """Very small in-memory TTL cache (process-local)."""
+
+    def __init__(self, maxsize: int = 10_000):
+        self._maxsize = maxsize
+        self._store: dict[str, _CacheEntry[T]] = {}
+
+    def get(self, key: str) -> Optional[T]:
+        e = self._store.get(key)
+        if e is None:
+            return None
+        if e.expires_at <= time.time():
+            self._store.pop(key, None)
+            return None
+        return e.value
+
+    def set(self, key: str, value: T, ttl_seconds: int) -> None:
+        if ttl_seconds <= 0:
+            return
+        if len(self._store) >= self._maxsize:
+            # naive eviction: drop expired, otherwise drop arbitrary oldest-like key
+            now = time.time()
+            for k in list(self._store.keys()):
+                if self._store[k].expires_at <= now:
+                    self._store.pop(k, None)
+            if len(self._store) >= self._maxsize:
+                self._store.pop(next(iter(self._store.keys())), None)
+
+        self._store[key] = _CacheEntry(value=value, expires_at=time.time() + ttl_seconds)

src/celine/dataset/api/dataset_query/row_filters/handlers/__init__.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from .direct_user_match import DirectUserMatchHandler
+from .http_in_list import HttpInListHandler
+from .table_pointer import TablePointerHandler
+from .rec_registry import RecRegistryHandler
+
+__all__ = [
+    "DirectUserMatchHandler",
+    "HttpInListHandler",
+    "TablePointerHandler",
+    "RecRegistryHandler",
+]