Skip to content

client

stac_fastapi.extensions.core.aggregation.client

Aggregation extensions clients.

AsyncBaseAggregationClient

Bases: ABC

Defines an async pattern for implementing the STAC aggregation extension.

Source code in stac_fastapi/extensions/core/aggregation/client.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@attr.s
class AsyncBaseAggregationClient(abc.ABC):
    """Defines an async pattern for implementing the STAC aggregation extension."""

    # BUCKET = Bucket
    # AGGREGAION = Aggregation
    # AGGREGATION_COLLECTION = AggregationCollection

    async def get_aggregations(
        self, collection_id: Optional[str] = None, **kwargs
    ) -> AggregationCollection:
        """Get the aggregations available for the given collection_id.

        If collection_id is None, returns the available aggregations over all
        collections.
        """
        return AggregationCollection(
            type="AggregationCollection",
            aggregations=[Aggregation(name="total_count", data_type="integer")],
            links=[
                {
                    "rel": "root",
                    "type": "application/json",
                    "href": "https://example.org/",
                },
                {
                    "rel": "self",
                    "type": "application/json",
                    "href": "https://example.org/aggregations",
                },
            ],
        )

    async def aggregate(
        self,
        collection_id: Optional[str] = None,
        aggregations: Optional[Union[str, List[str]]] = None,
        collections: Optional[List[str]] = None,
        ids: Optional[List[str]] = None,
        bbox: Optional[BBox] = None,
        intersects: Optional[Geometry] = None,
        datetime: Optional[DateTimeType] = None,
        limit: Optional[int] = 10,
        **kwargs,
    ) -> AggregationCollection:
        """Return the aggregation buckets for a given search result"""
        return AggregationCollection(
            type="AggregationCollection",
            aggregations=[],
            links=[
                {
                    "rel": "root",
                    "type": "application/json",
                    "href": "https://example.org/",
                },
                {
                    "rel": "self",
                    "type": "application/json",
                    "href": "https://example.org/aggregations",
                },
            ],
        )

aggregate async 

aggregate(
    collection_id: Optional[str] = None,
    aggregations: Optional[Union[str, List[str]]] = None,
    collections: Optional[List[str]] = None,
    ids: Optional[List[str]] = None,
    bbox: Optional[BBox] = None,
    intersects: Optional[Geometry] = None,
    datetime: Optional[DateTimeType] = None,
    limit: Optional[int] = 10,
    **kwargs
) -> AggregationCollection

Return the aggregation buckets for a given search result

Source code in stac_fastapi/extensions/core/aggregation/client.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
async def aggregate(
    self,
    collection_id: Optional[str] = None,
    aggregations: Optional[Union[str, List[str]]] = None,
    collections: Optional[List[str]] = None,
    ids: Optional[List[str]] = None,
    bbox: Optional[BBox] = None,
    intersects: Optional[Geometry] = None,
    datetime: Optional[DateTimeType] = None,
    limit: Optional[int] = 10,
    **kwargs,
) -> AggregationCollection:
    """Return the aggregation buckets for a given search result"""
    return AggregationCollection(
        type="AggregationCollection",
        aggregations=[],
        links=[
            {
                "rel": "root",
                "type": "application/json",
                "href": "https://example.org/",
            },
            {
                "rel": "self",
                "type": "application/json",
                "href": "https://example.org/aggregations",
            },
        ],
    )

get_aggregations async 

get_aggregations(collection_id: Optional[str] = None, **kwargs) -> AggregationCollection

Get the aggregations available for the given collection_id.

If collection_id is None, returns the available aggregations over all collections.

Source code in stac_fastapi/extensions/core/aggregation/client.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def get_aggregations(
    self, collection_id: Optional[str] = None, **kwargs
) -> AggregationCollection:
    """Get the aggregations available for the given collection_id.

    If collection_id is None, returns the available aggregations over all
    collections.
    """
    return AggregationCollection(
        type="AggregationCollection",
        aggregations=[Aggregation(name="total_count", data_type="integer")],
        links=[
            {
                "rel": "root",
                "type": "application/json",
                "href": "https://example.org/",
            },
            {
                "rel": "self",
                "type": "application/json",
                "href": "https://example.org/aggregations",
            },
        ],
    )

BaseAggregationClient

Bases: ABC

Defines a pattern for implementing the STAC aggregation extension.

Source code in stac_fastapi/extensions/core/aggregation/client.py 
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
@attr.s
class BaseAggregationClient(abc.ABC):
    """Defines a pattern for implementing the STAC aggregation extension."""

    # BUCKET = Bucket
    # AGGREGAION = Aggregation
    # AGGREGATION_COLLECTION = AggregationCollection

    def get_aggregations(
        self, collection_id: Optional[str] = None, **kwargs
    ) -> AggregationCollection:
        """Get the aggregations available for the given collection_id.

        If collection_id is None, returns the available aggregations over all
        collections.
        """
        return AggregationCollection(
            type="AggregationCollection",
            aggregations=[Aggregation(name="total_count", data_type="integer")],
            links=[
                {
                    "rel": "root",
                    "type": "application/json",
                    "href": "https://example.org/",
                },
                {
                    "rel": "self",
                    "type": "application/json",
                    "href": "https://example.org/aggregations",
                },
            ],
        )

    def aggregate(
        self, collection_id: Optional[str] = None, **kwargs
    ) -> AggregationCollection:
        """Return the aggregation buckets for a given search result"""
        return AggregationCollection(
            type="AggregationCollection",
            aggregations=[],
            links=[
                {
                    "rel": "root",
                    "type": "application/json",
                    "href": "https://example.org/",
                },
                {
                    "rel": "self",
                    "type": "application/json",
                    "href": "https://example.org/aggregations",
                },
            ],
        )

aggregate 

aggregate(collection_id: Optional[str] = None, **kwargs) -> AggregationCollection

Return the aggregation buckets for a given search result

Source code in stac_fastapi/extensions/core/aggregation/client.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def aggregate(
    self, collection_id: Optional[str] = None, **kwargs
) -> AggregationCollection:
    """Return the aggregation buckets for a given search result"""
    return AggregationCollection(
        type="AggregationCollection",
        aggregations=[],
        links=[
            {
                "rel": "root",
                "type": "application/json",
                "href": "https://example.org/",
            },
            {
                "rel": "self",
                "type": "application/json",
                "href": "https://example.org/aggregations",
            },
        ],
    )

get_aggregations 

get_aggregations(collection_id: Optional[str] = None, **kwargs) -> AggregationCollection

Get the aggregations available for the given collection_id.

If collection_id is None, returns the available aggregations over all collections.

Source code in stac_fastapi/extensions/core/aggregation/client.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def get_aggregations(
    self, collection_id: Optional[str] = None, **kwargs
) -> AggregationCollection:
    """Get the aggregations available for the given collection_id.

    If collection_id is None, returns the available aggregations over all
    collections.
    """
    return AggregationCollection(
        type="AggregationCollection",
        aggregations=[Aggregation(name="total_count", data_type="integer")],
        links=[
            {
                "rel": "root",
                "type": "application/json",
                "href": "https://example.org/",
            },
            {
                "rel": "self",
                "type": "application/json",
                "href": "https://example.org/aggregations",
            },
        ],
    )