Skip to content

Cql2ApplyFilterQueryStringMiddleware

Middleware to inject CQL2 filters into the query string for GET/list endpoints.

Cql2ApplyFilterQueryStringMiddleware dataclass

Middleware to inject CQL2 filters into the query string for GET/list endpoints.

Parameters:

Name Type Description Default
app Callable[list, Awaitable[None]]
required
state_key str
'cql2_filter'
Source code in src/stac_auth_proxy/middleware/Cql2ApplyFilterQueryStringMiddleware.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@required_conformance(
    r"http://www.opengis.net/spec/cql2/1.0/conf/basic-cql2",
    r"http://www.opengis.net/spec/cql2/1.0/conf/cql2-text",
    r"http://www.opengis.net/spec/cql2/1.0/conf/cql2-json",
)
@dataclass(frozen=True)
class Cql2ApplyFilterQueryStringMiddleware:
    """Middleware to inject CQL2 filters into the query string for GET/list endpoints."""

    app: ASGIApp
    state_key: str = "cql2_filter"

    single_record_endpoints = [
        r"^/collections/([^/]+)/items/([^/]+)$",
        r"^/collections/([^/]+)$",
    ]

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Apply the CQL2 filter to the query string."""
        if scope["type"] != "http":
            return await self.app(scope, receive, send)

        request = Request(scope)
        cql2_filter: Optional[Expr] = getattr(request.state, self.state_key, None)
        if not cql2_filter:
            return await self.app(scope, receive, send)

        # Only handle GET requests that are not single-record endpoints
        if request.method != "GET":
            return await self.app(scope, receive, send)
        if any(
            re.match(expr, request.url.path) for expr in self.single_record_endpoints
        ):
            return await self.app(scope, receive, send)

        # Inject filter into query string
        scope = dict(scope)
        scope["query_string"] = filters.append_qs_filter(request.url.query, cql2_filter)
        return await self.app(scope, receive, send)

__call__(scope: Scope, receive: Receive, send: Send) -> None async

Apply the CQL2 filter to the query string.

Source code in src/stac_auth_proxy/middleware/Cql2ApplyFilterQueryStringMiddleware.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    """Apply the CQL2 filter to the query string."""
    if scope["type"] != "http":
        return await self.app(scope, receive, send)

    request = Request(scope)
    cql2_filter: Optional[Expr] = getattr(request.state, self.state_key, None)
    if not cql2_filter:
        return await self.app(scope, receive, send)

    # Only handle GET requests that are not single-record endpoints
    if request.method != "GET":
        return await self.app(scope, receive, send)
    if any(
        re.match(expr, request.url.path) for expr in self.single_record_endpoints
    ):
        return await self.app(scope, receive, send)

    # Inject filter into query string
    scope = dict(scope)
    scope["query_string"] = filters.append_qs_filter(request.url.query, cql2_filter)
    return await self.app(scope, receive, send)