EPDD-13457 feat(filter): typed filter builder generated from the API filter grammar

.github/workflows/filter-builder-ci.yml

@@ -0,0 +1,22 @@
+name: Filter builder CI
+
+on:
+  push:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: "3.11"
+      - run: python -m pip install -e .
+      - run: python -m pip install "ruff==0.16.1"
+      - run: PYTHONPATH=src python -m unittest discover -s tests -p 'test_filter.py'
+      - run: python scripts/generate_filter_table.py tests/fixtures/filter-grammar.openapi.json
+      - run: git diff --exit-code -- src/profound/lib/filter_table.py

.github/workflows/sync-filter-table.yml

@@ -0,0 +1,41 @@
+name: Sync filter table
+
+on:
+  push:
+    branches: [scalar-next]
+    # The Scalar platform touches this manifest on every regeneration.
+    paths: [scalar-sdk.manifest.json]
+  schedule:
+    - cron: "17 6 * * *"
+  workflow_dispatch:
+
+permissions:
+  contents: write
+
+concurrency:
+  group: sync-filter-table
+  cancel-in-progress: false
+
+jobs:
+  sync:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
+        with:
+          ref: scalar-next
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: "3.11"
+      - run: python -m pip install "ruff==0.16.1"
+      - run: python scripts/generate_filter_table.py
+      - name: Commit changed filter table
+        run: |
+          if git diff --quiet -- src/profound/lib/filter_table.py; then
+            exit 0
+          fi
+          git config user.name "github-actions[bot]"
+          git config user.email "41898282+github-actions[bot]@users.noreply.github.com"
+          git add src/profound/lib/filter_table.py
+          git commit -m "fix(filter): sync filter table with the published API spec"
+          # Plain push: a lost race against a regeneration push fails loudly, and the next run re-converges.
+          git push origin HEAD:scalar-next

scripts/generate_filter_table.py

@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+REGISTRY_URL = "https://registry.scalar.com/@profound/apis/external-api/latest?format=json"
+OUTPUT = Path(__file__).parents[1] / "src" / "profound" / "lib" / "filter_table.py"
+
+
+def load_spec(source: str) -> dict[str, Any]:
+    if source.startswith(("http://", "https://")):
+        with urllib.request.urlopen(source) as response:
+            return json.load(response)
+    with Path(source).open(encoding="utf-8") as file:
+        return json.load(file)
+
+
+def generate(spec: dict[str, Any]) -> str:
+    grammar = spec.get("x-profound-filter-grammar")
+    if not grammar:
+        raise ValueError("spec has no x-profound-filter-grammar extension")
+
+    lines = [
+        "from __future__ import annotations",
+        "",
+        "from dataclasses import dataclass",
+        "from typing import Dict, Final, Literal, Tuple",
+        "",
+        "",
+        "@dataclass(frozen=True)",
+        "class FieldSpec:",
+        "    name: str",
+        '    layer: Literal["prompt", "entity"]',
+        "    ops: Tuple[str, ...]",
+        "",
+        "",
+        f"MAX_DEPTH: Final = {grammar['max_depth']}",
+        "",
+        "FIELD_TABLE: Final[Dict[str, FieldSpec]] = {",
+    ]
+    for name, field in grammar["fields"].items():
+        ops = ", ".join(repr(op) for op in field["ops"])
+        if len(field["ops"]) == 1:
+            ops += ","
+        lines.extend(
+            [
+                f"    {name!r}: FieldSpec({name!r}, {field['layer']!r}, ({ops})),",
+            ]
+        )
+    lines.extend(["}", "", "", "class Fields:"])
+    for name in grammar["fields"]:
+        lines.append(f"    {name}: Final = FIELD_TABLE[{name!r}]")
+    lines.append("")
+    return "\n".join(lines)
+
+
+def main() -> int:
+    source = sys.argv[1] if len(sys.argv) > 1 else REGISTRY_URL
+    try:
+        content = generate(load_spec(source))
+    except (OSError, ValueError, json.JSONDecodeError, urllib.error.URLError) as error:
+        print(f"failed to generate filter table: {error}", file=sys.stderr)
+        return 1
+    OUTPUT.write_text(
+        "# Generated by scripts/generate_filter_table.py from the x-profound-filter-grammar extension "
+        "of the Profound OpenAPI spec. Do not edit by hand.\n\n" + content,
+        encoding="utf-8",
+    )
+    subprocess.run(["ruff", "format", str(OUTPUT)], check=True)
+    print(f"wrote {OUTPUT}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/profound/lib/filter.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from typing import Dict, List, Optional, Set, Union
+
+from .filter_table import FIELD_TABLE, MAX_DEPTH, FieldSpec
+
+FilterNode = Dict[str, object]
+Value = Union[str, List[str]]
+
+
+def _depth(node: FilterNode) -> int:
+    if "and" in node:
+        return 1 + max(_depth(child) for child in node["and"])  # type: ignore[index]
+    if "or" in node:
+        return 1 + max(_depth(child) for child in node["or"])  # type: ignore[index]
+    if "not" in node:
+        return 1 + _depth(node["not"])  # type: ignore[arg-type]
+    return 1
+
+
+def _collect_layers(node: FilterNode, layers: Set[str]) -> None:
+    if "and" in node:
+        for child in node["and"]:  # type: ignore[union-attr]
+            _collect_layers(child, layers)
+    elif "or" in node:
+        for child in node["or"]:  # type: ignore[union-attr]
+            _collect_layers(child, layers)
+    elif "not" in node:
+        _collect_layers(node["not"], layers)  # type: ignore[arg-type]
+    else:
+        layers.add(FIELD_TABLE[node["field"]].layer)  # type: ignore[index]
+
+
+def _check_depth(node: FilterNode) -> FilterNode:
+    depth = _depth(node)
+    if depth > MAX_DEPTH:
+        raise ValueError(f"Filter tree exceeds the maximum nesting depth of {MAX_DEPTH} (got {depth})")
+    return node
+
+
+def _check_single_layer(node: FilterNode, kind: str) -> None:
+    layers: Set[str] = set()
+    _collect_layers(node, layers)
+    if len(layers) > 1:
+        raise ValueError(
+            f'Cannot mix prompt-layer and entity-layer fields under "{kind}"; '
+            'combine layers at the top level with "and"'
+        )
+
+
+def _leaf(field: FieldSpec, op: str, value: Optional[Value] = None) -> FilterNode:
+    spec = FIELD_TABLE.get(field.name)
+    if spec is None:
+        raise ValueError(f"Unknown field: {field.name}")
+    if op not in spec.ops:
+        raise ValueError(f'Field "{field.name}" does not support op "{op}" (allowed: {", ".join(spec.ops)})')
+    node: FilterNode = {"field": field.name, "op": op}
+    if value is not None:
+        node["value"] = value
+    return node
+
+
+def _list_leaf(field: FieldSpec, op: str, values: List[str]) -> FilterNode:
+    if not isinstance(values, list) or not values:
+        raise ValueError(f'"{op}" requires a non-empty list of values')
+    return _leaf(field, op, values)
+
+
+def and_(*nodes: FilterNode) -> FilterNode:
+    if not nodes:
+        raise ValueError('"and" requires at least one node')
+    return _check_depth({"and": list(nodes)})
+
+
+def or_(*nodes: FilterNode) -> FilterNode:
+    if not nodes:
+        raise ValueError('"or" requires at least one node')
+    node = _check_depth({"or": list(nodes)})
+    _check_single_layer(node, "or")
+    return node
+
+
+def not_(node: FilterNode) -> FilterNode:
+    checked = _check_depth({"not": node})
+    _check_single_layer(checked, "not")
+    return checked
+
+
+def equals(field: FieldSpec, value: str) -> FilterNode:
+    return _leaf(field, "is", value)
+
+
+def not_equals(field: FieldSpec, value: str) -> FilterNode:
+    return _leaf(field, "not_is", value)
+
+
+def in_(field: FieldSpec, values: List[str]) -> FilterNode:
+    return _list_leaf(field, "in", values)
+
+
+def not_in(field: FieldSpec, values: List[str]) -> FilterNode:
+    return _list_leaf(field, "not_in", values)
+
+
+def contains(field: FieldSpec, value: str) -> FilterNode:
+    return _leaf(field, "contains", value)
+
+
+def not_contains(field: FieldSpec, value: str) -> FilterNode:
+    return _leaf(field, "not_contains", value)
+
+
+def contains_insensitive(field: FieldSpec, value: str) -> FilterNode:
+    return _leaf(field, "contains_case_insensitive", value)
+
+
+def not_contains_insensitive(field: FieldSpec, value: str) -> FilterNode:
+    return _leaf(field, "not_contains_case_insensitive", value)
+
+
+def matches(field: FieldSpec, pattern: str) -> FilterNode:
+    if not isinstance(pattern, str) or len(pattern) < 3:
+        raise ValueError('"matches" requires a regex pattern of at least 3 characters')
+    return _leaf(field, "matches", pattern)
+
+
+def exists(field: FieldSpec) -> FilterNode:
+    return _leaf(field, "exists")