File: mixins.py

package info (click to toggle)
pystac-client 0.9.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 47,416 kB
  • sloc: python: 4,652; sh: 74; makefile: 60
file content (70 lines) | stat: -rw-r--r-- 2,214 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import warnings
from typing import Any

import pystac

from pystac_client._utils import urljoin
from pystac_client.conformance import ConformanceClasses
from pystac_client.exceptions import APIError
from pystac_client.stac_api_io import StacApiIO
from pystac_client.warnings import DoesNotConformTo, MissingLink

QUERYABLES_REL = "http://www.opengis.net/def/rel/ogc/1.0/queryables"
QUERYABLES_ENDPOINT = "queryables"


class StacAPIObject(pystac.STACObject):
    _stac_io: StacApiIO | None

    def conforms_to(self, conformance_class: str | ConformanceClasses) -> bool:
        raise NotImplementedError


class BaseMixin(StacAPIObject):
    def _get_href(self, rel: str, link: pystac.Link | None, endpoint: str) -> str:
        if link and isinstance(link.href, str):
            href = link.absolute_href
        else:
            warnings.warn(MissingLink(rel, self.__class__.__name__), stacklevel=2)
            href = urljoin(self.self_href, endpoint)
        return href


class QueryablesMixin(BaseMixin):
    """Mixin for adding support for /queryables endpoint"""

    def get_queryables_from(self, url: str) -> dict[str, Any]:
        """Return all queryables.

        Output is a dictionary that can be used in ``jsonshema.validate``

        Args:
            url: a queryables url

        Return:
            Dict[str, Any]: Dictionary containing queryable fields
        """

        if self._stac_io is None:
            raise APIError("API access is not properly configured")

        result = self._stac_io.read_json(url)
        if "properties" not in result:
            raise APIError(
                f"Invalid response from {QUERYABLES_ENDPOINT}: "
                "expected 'properties' attribute"
            )

        return result

    def get_queryables(self) -> dict[str, Any]:
        url = self._get_queryables_href()
        return self.get_queryables_from(url)

    def _get_queryables_href(self) -> str:
        if not self.conforms_to(ConformanceClasses.FILTER):
            raise DoesNotConformTo(ConformanceClasses.FILTER.name)

        link = self.get_single_link(QUERYABLES_REL)
        href = self._get_href(QUERYABLES_REL, link, QUERYABLES_ENDPOINT)
        return href