stac_auth_proxy.filters

CQL2 filter factories.

Opa dataclass

Call Open Policy Agent (OPA) to generate CQL2 filters from request context.

Parameters:

Name	Type	Description	Default
host	str		required
decision	str		required

Attributes:

Name	Type	Description
client	AsyncClient

Source code in src/stac_auth_proxy/filters/opa.py

@dataclass
class Opa:
    """Call Open Policy Agent (OPA) to generate CQL2 filters from request context."""

    host: str
    decision: str

    client: httpx.AsyncClient = field(init=False)

    def __post_init__(self):
        """Initialize the client."""
        self.client = httpx.AsyncClient(base_url=self.host)

    async def __call__(self, context: dict[str, Any]) -> str:
        """
        Generate a CQL2 filter for the request.

        Not cached: the policy may depend on any part of the request (method,
        path, ...), so a result is only valid for the request it was computed for.
        """
        response = await self.client.post(
            f"/v1/data/{self.decision}",
            json={"input": context},
        )
        return response.raise_for_status().json()["result"]

__call__(context: dict[str, Any]) -> str async

Generate a CQL2 filter for the request.

Not cached: the policy may depend on any part of the request (method, path, ...), so a result is only valid for the request it was computed for.

Source code in src/stac_auth_proxy/filters/opa.py

async def __call__(self, context: dict[str, Any]) -> str:
    """
    Generate a CQL2 filter for the request.

    Not cached: the policy may depend on any part of the request (method,
    path, ...), so a result is only valid for the request it was computed for.
    """
    response = await self.client.post(
        f"/v1/data/{self.decision}",
        json={"input": context},
    )
    return response.raise_for_status().json()["result"]

__post_init__()

Initialize the client.

Source code in src/stac_auth_proxy/filters/opa.py

def __post_init__(self):
    """Initialize the client."""
    self.client = httpx.AsyncClient(base_url=self.host)

Template dataclass

Generate CQL2 filter expressions via Jinja2 templating.

Parameters:

Name	Type	Description	Default
template_str	str		required

Attributes:

Name	Type	Description
env	SandboxedEnvironment

Source code in src/stac_auth_proxy/filters/template.py

@dataclass
class Template:
    """Generate CQL2 filter expressions via Jinja2 templating."""

    template_str: str
    env: SandboxedEnvironment = field(init=False)

    def __post_init__(self):
        """Initialize the Jinja2 environment."""
        self.env = SandboxedEnvironment(loader=BaseLoader).from_string(
            self.template_str
        )

    async def __call__(self, context: dict[str, Any]) -> str:
        """Render a CQL2 filter expression with the request and auth token."""
        return self.env.render(**context).strip()

__call__(context: dict[str, Any]) -> str async

Render a CQL2 filter expression with the request and auth token.

Source code in src/stac_auth_proxy/filters/template.py

async def __call__(self, context: dict[str, Any]) -> str:
    """Render a CQL2 filter expression with the request and auth token."""
    return self.env.render(**context).strip()

__post_init__()

Initialize the Jinja2 environment.

Source code in src/stac_auth_proxy/filters/template.py

def __post_init__(self):
    """Initialize the Jinja2 environment."""
    self.env = SandboxedEnvironment(loader=BaseLoader).from_string(
        self.template_str
    )