feat(records): aggregate endpoint

cognite/client/_api/data_modeling/records.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Literal
 
 from cognite.client._api_client import APIClient
@@ -11,13 +11,15 @@
     RecordId,
     RecordIdSequence,
     RecordList,
+    RecordsAggregation,
     RecordSourceSelector,
     RecordTargetUnit,
     RecordTargetUnits,
     RecordWrite,
     SyncRecord,
     SyncRecordList,
     TimeRange,
+    _dump_aggregate_value,
 )
 from cognite.client.data_classes.filters import Filter
 from cognite.client.utils._experimental import FeaturePreviewWarning
@@ -235,6 +237,106 @@ async def upsert(
             no_response=True,
         )
 
+    async def aggregate(
+        self,
+        aggregates: Mapping[str, Any],
+        *,
+        stream_id: str,
+        last_updated_time: TimeRange | None = None,
+        filter: Filter | dict[str, Any] | None = None,
+        target_units: RecordTargetUnits | Sequence[RecordTargetUnit] | None = None,
+        include_typing: bool = False,
+    ) -> RecordsAggregation:
+        """`Aggregate records from a stream <https://api-docs.cognite.com/20230101/tag/Records/operation/aggregateRecords>`_.
+
+        Args:
+            aggregates (Mapping[str, Any]): Aggregate request tree keyed by client-defined aggregate IDs.
+            stream_id (str): External ID of the stream to aggregate from.
+            last_updated_time (TimeRange | None): Filter records by last-updated time.
+                **Required** for immutable streams (must include a lower bound).
+            filter (Filter | dict[str, Any] | None): Filter expression.
+            target_units (RecordTargetUnits | Sequence[RecordTargetUnit] | None): Unit conversion specification.
+            include_typing (bool): Include property type metadata in the response.
+
+        Returns:
+            RecordsAggregation: Aggregate results keyed by the requested aggregate IDs.
+
+        Examples:
+
+            Aggregate average temperature using typed helpers:
+
+                >>> from cognite.client import CogniteClient
+                >>> from cognite.client.data_classes.data_modeling.records import Avg
+                >>> client = CogniteClient()
+                >>> res = client.data_modeling.records.aggregate(
+                ...     stream_id="my-stream",
+                ...     aggregates={
+                ...         "avg_temp": Avg(["my-space", "sensor", "temperature"]),
+                ...     },
+                ... )
+
+            Count records with a time range filter:
+
+                >>> from cognite.client.data_classes.data_modeling.records import Count, TimeRange
+                >>> res = client.data_modeling.records.aggregate(
+                ...     stream_id="my-stream",
+                ...     aggregates={"total": Count()},
+                ...     last_updated_time=TimeRange(gte=1705341600000),
+                ... )
+
+            Daily time histogram with nested metric aggregates:
+
+                >>> from cognite.client.data_classes.data_modeling.records import (
+                ...     Max,
+                ...     Min,
+                ...     TimeHistogram,
+                ... )
+                >>> res = client.data_modeling.records.aggregate(
+                ...     stream_id="my-stream",
+                ...     aggregates={
+                ...         "by_day": TimeHistogram(
+                ...             ["my-space", "sensor", "timestamp"],
+                ...             calendar_interval="1d",
+                ...             aggregates={
+                ...                 "min_temp": Min(["my-space", "sensor", "temperature"]),
+                ...                 "max_temp": Max(["my-space", "sensor", "temperature"]),
+                ...             },
+                ...         ),
+                ...     },
+                ... )
+
+            Unique values with sub-aggregation:
+
+                >>> from cognite.client.data_classes.data_modeling.records import Sum, UniqueValues
+                >>> res = client.data_modeling.records.aggregate(
+                ...     stream_id="my-stream",
+                ...     aggregates={
+                ...         "by_region": UniqueValues(
+                ...             ["my-space", "sensor", "region"],
+                ...             aggregates={"total_output": Sum(["my-space", "sensor", "output"])},
+                ...             size=10,
+                ...         ),
+                ...     },
+                ... )
+        """
+        self._warning.warn()
+        body: dict[str, Any] = {"aggregates": _dump_aggregate_value(aggregates)}
+        if last_updated_time is not None:
+            body["lastUpdatedTime"] = last_updated_time.dump()
+        if filter is not None:
+            body["filter"] = filter.dump() if isinstance(filter, Filter) else filter
+        if target_units is not None:
+            body["targetUnits"] = self._dump_target_units(target_units)
+        if include_typing:
+            body["includeTyping"] = True
+
+        res = await self._post(
+            url_path=self._records_url(stream_id, "/aggregate"),
+            json=body,
+            semaphore=self._get_semaphore("read"),
+        )
+        return RecordsAggregation._load(res.json())
+
     async def filter(
         self,
         stream_id: str,

cognite/client/data_classes/data_modeling/__init__.py

@@ -120,19 +120,39 @@
     UnionAll,
 )
 from cognite.client.data_classes.data_modeling.records import (
+    Avg,
+    Count,
+    FilterAggregateResult,
+    Filters,
+    Max,
+    MetricAggregateResult,
+    Min,
+    MovingFunction,
+    MovingFunctionAggregateResult,
+    NumberHistogram,
+    NumberHistogramAggregateResult,
     Record,
     RecordContainerId,
     RecordId,
     RecordList,
+    RecordsAggregate,
+    RecordsAggregateResult,
+    RecordsAggregation,
+    RecordsBucket,
     RecordSource,
     RecordSourceSelector,
     RecordTargetUnit,
     RecordTargetUnits,
     RecordWrite,
     RecordWriteList,
+    Sum,
     SyncRecord,
     SyncRecordList,
+    TimeHistogram,
+    TimeHistogramAggregateResult,
     TimeRange,
+    UniqueValues,
+    UniqueValuesAggregateResult,
 )
 from cognite.client.data_classes.data_modeling.spaces import Space, SpaceApply, SpaceApplyList, SpaceList
 from cognite.client.data_classes.data_modeling.streams import (
@@ -169,6 +189,7 @@
 __all__ = [
     "AggregatedValue",
     "Aggregation",
+    "Avg",
     "BTreeIndex",
     "BTreeIndexApply",
     "Boolean",
@@ -185,6 +206,7 @@
     "ContainerProperty",
     "ContainerPropertyApply",
     "ContainerUsedFor",
+    "Count",
     "DataModel",
     "DataModelApply",
     "DataModelApplyList",
@@ -215,6 +237,8 @@
     "ExecutionPlan",
     "FileReference",
     "Filter",
+    "FilterAggregateResult",
+    "Filters",
     "Float32",
     "Float64",
     "Index",
@@ -235,6 +259,11 @@
     "Json",
     "MappedProperty",
     "MappedPropertyApply",
+    "Max",
+    "MetricAggregateResult",
+    "Min",
+    "MovingFunction",
+    "MovingFunctionAggregateResult",
     "MultiEdgeConnection",
     "MultiEdgeConnectionApply",
     "MultiReverseDirectRelation",
@@ -251,6 +280,8 @@
     "NodeOrEdgeResultSetExpression",
     "NodeResultSetExpression",
     "NodeResultSetExpressionSync",
+    "NumberHistogram",
+    "NumberHistogramAggregateResult",
     "PropertyId",
     "PropertyOptions",
     "PropertyType",
@@ -267,6 +298,10 @@
     "RecordTargetUnits",
     "RecordWrite",
     "RecordWriteList",
+    "RecordsAggregate",
+    "RecordsAggregateResult",
+    "RecordsAggregation",
+    "RecordsBucket",
     "RequiresConstraint",
     "RequiresConstraintApply",
     "ResultSetExpression",
@@ -291,9 +326,12 @@
     "StreamTemplateWriteSettings",
     "StreamWrite",
     "SubscriptionContext",
+    "Sum",
     "SyncRecord",
     "SyncRecordList",
     "Text",
+    "TimeHistogram",
+    "TimeHistogramAggregateResult",
     "TimeRange",
     "TimeSeriesReference",
     "Timestamp",
@@ -304,6 +342,8 @@
     "TypedNodeApply",
     "Union",
     "UnionAll",
+    "UniqueValues",
+    "UniqueValuesAggregateResult",
     "UniquenessConstraint",
     "UniquenessConstraintApply",
     "VersionedDataModelingId",