Passing tests

Author: alukach

src/stac_auth_proxy/app.py

@@ -6,16 +6,16 @@
 """
 
 import logging
-from typing import Optional, Annotated
+from typing import Optional
 
-from fastapi import FastAPI, Security, Request, Depends
-from cql2 import Expr
+from fastapi import FastAPI, Security
 
 from .auth import OpenIdConnectAuth
 from .config import Settings
 from .handlers import OpenApiSpecHandler, ReverseProxyHandler
 from .middleware import AddProcessTimeHeaderMiddleware
-from .utils import apply_filter
+
+# from .utils import apply_filter
 
 logger = logging.getLogger(__name__)
 

src/stac_auth_proxy/handlers/reverse_proxy.py

@@ -3,16 +3,16 @@
 import logging
 import time
 from dataclasses import dataclass
-from typing import Optional, Annotated
+from typing import Annotated, Optional
 
-from cql2 import Expr
 import httpx
-from fastapi import Request, Depends
+from cql2 import Expr
+from fastapi import Depends, Request
 from starlette.background import BackgroundTask
 from starlette.datastructures import MutableHeaders
 from starlette.responses import StreamingResponse
 
-from ..utils import update_qs
+from .. import utils
 
 logger = logging.getLogger(__name__)
 
@@ -33,18 +33,18 @@ def __post_init__(self):
             timeout=httpx.Timeout(timeout=15.0),
         )
 
-        self.proxy_request.__annotations__["collections_filter"] = Annotated[
-            Optional[Expr], Depends(self.collections_filter.dependency)
-        ]
-        self.stream.__annotations__["collections_filter"] = Annotated[
-            Optional[Expr], Depends(self.collections_filter.dependency)
-        ]
+        # Update annotations to support FastAPI's dependency injection
+        for endpoint in [self.proxy_request, self.stream]:
+            endpoint.__annotations__["collections_filter"] = Annotated[
+                Optional[Expr],
+                Depends(getattr(self.collections_filter, "dependency", lambda: None)),
+            ]
 
     async def proxy_request(
         self,
         request: Request,
         *,
-        collections_filter: Annotated[Optional[Expr], Depends(...)],
+        collections_filter: Annotated[Optional[Expr], Depends(...)] = None,
         stream=False,
     ) -> httpx.Response:
         """Proxy a request to the upstream STAC API."""
@@ -53,24 +53,22 @@ async def proxy_request(
         headers.setdefault("X-Forwarded-Host", request.url.hostname)
 
         path = request.url.path
-        query = request.url.query.encode("utf-8")
+        query = request.url.query
 
-        # https://github.com/fastapi/fastapi/discussions/7382#discussioncomment-5136466
-        # TODO: Examine filters
-        if collections_filter:
+        if utils.is_collection_endpoint(path) and collections_filter:
             if request.method == "GET" and path == "/collections":
-                query += b"&" + update_qs(
-                    request.query_params, filter=collections_filter.to_text()
-                )
-
-        url = httpx.URL(
-            path=path,
-            query=query,
-        )
+                query = utils.insert_filter(qs=query, filter=collections_filter)
+        elif utils.is_item_endpoint(path) and self.items_filter:
+            if request.method == "GET":
+                query = utils.insert_filter(qs=query, filter=self.items_filter)
 
+        # https://github.com/fastapi/fastapi/discussions/7382#discussioncomment-5136466
         rp_req = self.client.build_request(
             request.method,
-            url=url,
+            url=httpx.URL(
+                path=path,
+                query=query.encode("utf-8"),
+            ),
             headers=headers,
             content=request.stream(),
         )

src/stac_auth_proxy/utils.py

@@ -1,12 +1,10 @@
 """Utility functions."""
 
 import re
-from urllib.parse import urlparse
+from urllib.parse import parse_qs, urlencode, urlparse
 
 from cql2 import Expr
-from fastapi import Request
 from fastapi.dependencies.models import Dependant
-from starlette.datastructures import QueryParams
 from httpx import Headers
 
 
@@ -47,45 +45,28 @@ def has_any_security_requirements(dependency: Dependant) -> bool:
     )
 
 
-async def apply_filter(request: Request, filter: Expr) -> Request:
-    """Apply a CQL2 filter to a request."""
-    req_filter = request.query_params.get("filter") or (
-        (await request.json()).get("filter")
-        if request.headers.get("content-length")
-        else None
-    )
+def insert_filter(qs: str, filter: Expr) -> str:
+    """Insert a filter expression into a query string. If a filter already exists, combine them."""
+    qs_dict = parse_qs(qs)
 
-    new_filter = Expr(" AND ".join(e.to_text() for e in [req_filter, filter] if e))
-    new_filter.validate()
-
-    if request.method == "GET":
-        updated_scope = request.scope.copy()
-        updated_scope["query_string"] = update_qs(
-            request.query_params,
-            filter=new_filter.to_text(),
-        )
-        return Request(
-            scope=updated_scope,
-            receive=request.receive,
-            # send=request._send,
-        )
-
-    # TODO: Support POST/PUT/PATCH
-    # elif request.method == "POST":
-    #     request_body = await request.body()
-    #     query = request.url.query
-    #     query += "&" if query else "?"
-    #     query += f"filter={filter}"
-    #     request.url.query = query
-
-    return request
-
-
-def update_qs(query_params: QueryParams, **updates) -> bytes:
-    query_dict = {
-        **query_params,
-        **updates,
-    }
-    return "&".join(f"{key}={value}" for key, value in query_dict.items()).encode(
-        "utf-8"
-    )
+    filters = [Expr(f) for f in qs_dict.get("filter", [])]
+    filters.append(filter)
+
+    combined_filter = Expr(" AND ".join(e.to_text() for e in filters))
+    combined_filter.validate()
+
+    qs_dict["filter"] = [combined_filter.to_text()]
+
+    return urlencode(qs_dict, doseq=True)
+
+
+def is_collection_endpoint(path: str) -> bool:
+    """Check if the path is a collection endpoint."""
+    # TODO: Expand this to cover all cases where a collection filter should be applied
+    return path == "/collections"
+
+
+def is_item_endpoint(path: str) -> bool:
+    """Check if the path is an item endpoint."""
+    # TODO: Expand this to cover all cases where an item filter should be applied
+    return path == "/collection/{collection_id}/items"