Skip to content

aggregation

stac_fastapi.extensions.aggregation.aggregation

Aggregation Extension.

AggregationConformanceClasses

Bases: StrEnum

Conformance classes for the Aggregation extension.

See stac-api-extensions/aggregation

Source code in stac_fastapi/extensions/stac_fastapi/extensions/aggregation/aggregation.py 
18
19
20
21
22
23
24
25
class AggregationConformanceClasses(StrEnum):
    """Conformance classes for the Aggregation extension.

    See
    https://github.com/stac-api-extensions/aggregation
    """

    AGGREGATION = "https://api.stacspec.org/v0.3.0/aggregation"

AggregationExtension

Bases: ApiExtension

Aggregation Extension.

The purpose of the Aggregation Extension is to provide an endpoint similar to the Search endpoint (/search), but which will provide aggregated information on matching Items rather than the Items themselves. This is highly influenced by the Elasticsearch and OpenSearch aggregation endpoint, but with a more regular structure for responses.

The Aggregation extension adds several endpoints which allow the retrieval of available aggregation fields and aggregation buckets based on a seearch query: GET /aggregations POST /aggregations GET /collections/{collection_id}/aggregations POST /collections/{collection_id}/aggregations GET /aggregate POST /aggregate GET /collections/{collection_id}/aggregate POST /collections/{collection_id}/aggregate

github.com/stac-api-extensions/aggregation/blob/main/README.md

Attributes:

  • conformance_classes (list[str]) –

    Conformance classes provided by the extension

Source code in stac_fastapi/extensions/stac_fastapi/extensions/aggregation/aggregation.py 
 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
 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
@attr.s
class AggregationExtension(ApiExtension):
    """Aggregation Extension.

    The purpose of the Aggregation Extension is to provide an endpoint similar to
    the Search endpoint (/search), but which will provide aggregated information
    on matching Items rather than the Items themselves. This is highly influenced
    by the Elasticsearch and OpenSearch aggregation endpoint, but with a more
    regular structure for responses.

    The Aggregation extension adds several endpoints which allow the retrieval of
    available aggregation fields and aggregation buckets based on a seearch query:
        GET /aggregations
        POST /aggregations
        GET /collections/{collection_id}/aggregations
        POST /collections/{collection_id}/aggregations
        GET /aggregate
        POST /aggregate
        GET /collections/{collection_id}/aggregate
        POST /collections/{collection_id}/aggregate

    https://github.com/stac-api-extensions/aggregation/blob/main/README.md

    Attributes:
        conformance_classes: Conformance classes provided by the extension
    """

    GET: type[APIRequest] = AggregationExtensionGetRequest
    POST: type[BaseModel] = AggregationExtensionPostRequest

    client: AsyncBaseAggregationClient | BaseAggregationClient = attr.ib(
        factory=BaseAggregationClient
    )

    conformance_classes: list[str] = attr.ib(
        default=[AggregationConformanceClasses.AGGREGATION]
    )
    router: APIRouter = attr.ib(factory=APIRouter)

    def register(self, app: FastAPI) -> None:
        """Register the extension with a FastAPI application.

        Args:
            app: target FastAPI application.

        Returns:
            None
        """
        self.router.prefix = app.state.router_prefix
        self.router.add_api_route(
            name="Aggregations",
            path="/aggregations",
            methods=["GET", "POST"],
            endpoint=create_async_endpoint(self.client.get_aggregations, EmptyRequest),
        )
        self.router.add_api_route(
            name="Collection Aggregations",
            path="/collections/{collection_id}/aggregations",
            methods=["GET", "POST"],
            endpoint=create_async_endpoint(self.client.get_aggregations, CollectionUri),
        )
        self.router.add_api_route(
            name="Aggregate",
            path="/aggregate",
            methods=["GET"],
            endpoint=create_async_endpoint(self.client.aggregate, self.GET),
        )
        self.router.add_api_route(
            name="Aggregate",
            path="/aggregate",
            methods=["POST"],
            endpoint=create_async_endpoint(self.client.aggregate, self.POST),
        )
        self.router.add_api_route(
            name="Collection Aggregate",
            path="/collections/{collection_id}/aggregate",
            methods=["GET"],
            endpoint=create_async_endpoint(self.client.aggregate, self.GET),
        )
        self.router.add_api_route(
            name="Collection Aggregate",
            path="/collections/{collection_id}/aggregate",
            methods=["POST"],
            endpoint=create_async_endpoint(self.client.aggregate, self.POST),
        )
        app.include_router(self.router, tags=["Aggregation Extension"])

register 

register(app: FastAPI) -> None

Register the extension with a FastAPI application.

Parameters:

  • app (FastAPI) –

    target FastAPI application.

Returns:

  • None –

    None

Source code in stac_fastapi/extensions/stac_fastapi/extensions/aggregation/aggregation.py 
 67
 68
 69
 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
def register(self, app: FastAPI) -> None:
    """Register the extension with a FastAPI application.

    Args:
        app: target FastAPI application.

    Returns:
        None
    """
    self.router.prefix = app.state.router_prefix
    self.router.add_api_route(
        name="Aggregations",
        path="/aggregations",
        methods=["GET", "POST"],
        endpoint=create_async_endpoint(self.client.get_aggregations, EmptyRequest),
    )
    self.router.add_api_route(
        name="Collection Aggregations",
        path="/collections/{collection_id}/aggregations",
        methods=["GET", "POST"],
        endpoint=create_async_endpoint(self.client.get_aggregations, CollectionUri),
    )
    self.router.add_api_route(
        name="Aggregate",
        path="/aggregate",
        methods=["GET"],
        endpoint=create_async_endpoint(self.client.aggregate, self.GET),
    )
    self.router.add_api_route(
        name="Aggregate",
        path="/aggregate",
        methods=["POST"],
        endpoint=create_async_endpoint(self.client.aggregate, self.POST),
    )
    self.router.add_api_route(
        name="Collection Aggregate",
        path="/collections/{collection_id}/aggregate",
        methods=["GET"],
        endpoint=create_async_endpoint(self.client.aggregate, self.GET),
    )
    self.router.add_api_route(
        name="Collection Aggregate",
        path="/collections/{collection_id}/aggregate",
        methods=["POST"],
        endpoint=create_async_endpoint(self.client.aggregate, self.POST),
    )
    app.include_router(self.router, tags=["Aggregation Extension"])