refactor: expect filter generators to return string or dict, not Expr

Author: alukach

src/stac_auth_proxy/filters/template.py

@@ -3,7 +3,6 @@
 from dataclasses import dataclass, field
 from typing import Any
 
-from cql2 import Expr
 from jinja2 import BaseLoader, Environment
 
 
@@ -18,10 +17,6 @@ def __post_init__(self):
         """Initialize the Jinja2 environment."""
         self.env = Environment(loader=BaseLoader).from_string(self.template_str)
 
-    async def __call__(self, context: dict[str, Any]) -> Expr:
+    async def __call__(self, context: dict[str, Any]) -> str:
         """Render a CQL2 filter expression with the request and auth token."""
-        # TODO: How to handle the case where auth_token is null?
-        cql2_str = self.env.render(**context).strip()
-        cql2_expr = Expr(cql2_str)
-        cql2_expr.validate()
-        return cql2_expr
+        return self.env.render(**context).strip()

src/stac_auth_proxy/middleware/BuildCql2FilterMiddleware.py

@@ -3,7 +3,7 @@
 import json
 import re
 from dataclasses import dataclass
-from typing import Callable, Optional
+from typing import Any, Awaitable, Callable, Optional
 
 from cql2 import Expr
 from starlette.requests import Request
@@ -37,7 +37,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
         async def set_filter(body: Optional[dict] = None) -> None:
             assert filter_builder is not None
-            cql2_filter = await filter_builder(
+            filter_expr = await filter_builder(
                 {
                     "req": {
                         "path": request.url.path,
@@ -50,6 +50,7 @@ async def set_filter(body: Optional[dict] = None) -> None:
                     **scope["state"],
                 }
             )
+            cql2_filter = Expr(filter_expr)
             cql2_filter.validate()
             setattr(request.state, self.state_key, cql2_filter)
 
@@ -76,7 +77,9 @@ async def receive_build_filter() -> Message:
 
         return await self.app(scope, receive_build_filter, send)
 
-    def _get_filter(self, path: str) -> Optional[Callable[..., Expr]]:
+    def _get_filter(
+        self, path: str
+    ) -> Optional[Callable[..., Awaitable[str | dict[str, Any]]]]:
         """Get the CQL2 filter builder for the given path."""
         endpoint_filters = [
             (r"^/collections(/[^/]+)?$", self.collections_filter),