Skip to content

feat(records): aggregate endpoint #2705

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
andersfylling wants to merge 24 commits into
base: master
Choose a base branch
from
Open

feat(records): aggregate endpoint #2705

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
66221d4
feat(records): add aggregate endpoint
andersfylling Jun 25, 2026
dd98ac8
feat(records): add aggregate helpers
andersfylling Jun 25, 2026
6653552
Merge origin/master into aggregate branch
andersfylling Jun 30, 2026
389eee2
fix: import ordering and docstring example for aggregate
andersfylling Jun 30, 2026
119f28d
fix: import ordering in API file and exclude aggregate classes from g…
andersfylling Jun 30, 2026
b353db4
Merge branch 'master' into andersfylling/records/aggregate-pr1
andersfylling Jun 30, 2026
ce4149c
feat(records): add typed aggregate examples to docstring and exclude …
andersfylling Jun 30, 2026
fb9c35e
refactor(records): expose bucket aggregate buckets as a read-only pro…
andersfylling Jul 6, 2026
d46279c
Merge branch 'master' into andersfylling/records/aggregate-pr1
andersfylling Jul 12, 2026
a607d83
Update cognite/client/data_classes/data_modeling/records.py
andersfylling Jul 20, 2026
12524ec
update code example
andersfylling Jul 20, 2026
6c37d01
dont use sequence[str]
andersfylling Jul 20, 2026
100c3c1
unroll loops for aggregate results
andersfylling Jul 21, 2026
23d7777
split up responsibilities of init/load/dump
andersfylling Jul 21, 2026
bed39fb
remove redundant load/dump test
andersfylling Jul 21, 2026
4b3d786
commit to using proper types for aggs
andersfylling Jul 21, 2026
4b4bf0a
rewrite aggs
andersfylling Jul 21, 2026
a209fb3
cleanup the aggs examples
andersfylling Jul 21, 2026
aedf329
prefix records agg result types with Records
andersfylling Jul 21, 2026
9935753
Merge branch 'master' into andersfylling/records/aggregate-pr1
andersfylling Jul 21, 2026
04309fe
whops, re-add load support to the agg request types
andersfylling Jul 21, 2026
a0f7ab7
add unknown aggregate type
andersfylling Jul 21, 2026
26cb596
stricter types
andersfylling Jul 21, 2026
6385ad3
rename Avg => Average
andersfylling Jul 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
145 changes: 144 additions & 1 deletion cognite/client/_api/data_modeling/records.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -11,13 +11,16 @@
RecordId,
RecordIdSequence,
RecordList,
RecordsAggregate,
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
Expand Down Expand Up @@ -235,6 +238,146 @@ async def upsert(
no_response=True,
)

async def aggregate(
self,
aggregates: Mapping[str, RecordsAggregate | dict[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, RecordsAggregate | dict[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:

The examples below aggregate over a stream of padel game statistics records; each
example builds on the previous one.

The property paths used below:

>>> game_time = ["paddle", "game_statistics", "game_time"]
>>> player_name = ["paddle", "game_statistics", "player_name"]
>>> points_scored = ["paddle", "game_statistics", "points_scored"]


Find the average points scored across all games, using a typed helper:

>>> from cognite.client import CogniteClient
>>> from cognite.client.data_classes.data_modeling.records import Average
>>> client = CogniteClient()
>>> res = client.data_modeling.records.aggregate(
... stream_id="my-stream",
... aggregates={"avg_points_scored": Average(points_scored)},
... )


Count the total number of games, and how many of them have a recorded score, only
considering games updated after a given time:

>>> from cognite.client.data_classes.data_modeling.records import Count, TimeRange
>>> res = client.data_modeling.records.aggregate(
... stream_id="my-stream",
... aggregates={
... "total_games": Count(),
... "games_with_score": Count(points_scored),
... },
... last_updated_time=TimeRange(gt=1759276800000),
... )


Group games by day, then by player, and for each player-day compute their total,
highest, and average points scored, alongside the single highest score across all
games:

>>> from cognite.client.data_classes.data_modeling.records import (
... Average,
... Max,
... Sum,
... TimeHistogram,
... UniqueValues,
... )
>>> res = client.data_modeling.records.aggregate(
... stream_id="my-stream",
... aggregates={
... "my_groups_by_1d_range": TimeHistogram(
... property=game_time,
... calendar_interval="1d",
... aggregates={
... "my_groups_by_player_name": UniqueValues(
... property=player_name,
... aggregates={
... "my_player_daily_scores_sum": Sum(points_scored),
... "my_player_daily_scores_maximum": Max(points_scored),
... },
... ),
... "my_daily_scores_average": Average(points_scored),
... },
... ),
... "my_scores_maximum_across_all_games": Max(points_scored),
... },
... )


Bucket games by day and smooth the daily count with a 7-day moving average, using the
``MovingFunctions`` enum so the pipeline function name cannot be mistyped:

>>> from cognite.client.data_classes.data_modeling.records import (
... Count,
... MovingFunction,
... MovingFunctions,
... TimeHistogram,
... )
>>> res = client.data_modeling.records.aggregate(
... stream_id="my-stream",
... aggregates={
... "games_per_day": TimeHistogram(
... property=game_time,
... calendar_interval="1d",
... aggregates={
... "games": Count(),
... "games_7d_avg": MovingFunction(
... buckets_path="games",
... window=7,
... function=MovingFunctions.UNWEIGHTED_AVG,
... ),
... },
... ),
... },
... )
"""
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,
Expand Down
142 changes: 139 additions & 3 deletions cognite/client/_sync_api/data_modeling/records.py
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

20 changes: 20 additions & 0 deletions cognite/client/data_classes/data_modeling/__init__.py
View file
Open in desktop

@haakonvt haakonvt Jul 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should keep these out of cognite/client/data_classes/data_modeling/__init__.py and only in ...records.py

@andersfylling andersfylling Jul 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just a heads up it will be inconsistent with the other data classes that are accessible there. Do you want purely aggregate related classes, or all records/streams classes to be in their own module?

@andersfylling andersfylling Jul 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've prefixed the response agg types with records, and then i moved the builder types into the records module. Do you want me to move the response types into records as well?

Original file line number Diff line number Diff line change
Expand Up @@ -124,8 +124,18 @@
RecordContainerId,
RecordId,
RecordList,
RecordsAggregateResult,
RecordsAggregation,
RecordsBucket,
RecordsFilterAggregateResult,
RecordsMetricAggregateResult,
RecordsMovingFunctionAggregateResult,
RecordsNumberHistogramAggregateResult,
RecordSource,
RecordSourceSelector,
RecordsTimeHistogramAggregateResult,
RecordsUniqueValuesAggregateResult,
RecordsUnknownAggregateResult,
RecordTargetUnit,
RecordTargetUnits,
RecordWrite,
Expand Down Expand Up @@ -267,6 +277,16 @@
"RecordTargetUnits",
"RecordWrite",
"RecordWriteList",
"RecordsAggregateResult",
"RecordsAggregation",
"RecordsBucket",
"RecordsFilterAggregateResult",
"RecordsMetricAggregateResult",
"RecordsMovingFunctionAggregateResult",
"RecordsNumberHistogramAggregateResult",
"RecordsTimeHistogramAggregateResult",
"RecordsUniqueValuesAggregateResult",
"RecordsUnknownAggregateResult",
"RequiresConstraint",
"RequiresConstraintApply",
"ResultSetExpression",
Expand Down
Loading
Loading