roboto.domain.metrics.metric

Module Contents

class roboto.domain.metrics.metric.BulkPublishMetricsResult

Result of a bulk metric publish — may contain both successes and per-item failures.

failed: list[roboto.domain.metrics.record.PublishMetricsError]

succeeded: list[Metric]

class roboto.domain.metrics.metric.Metric(record, roboto_client)

A summary value recorded for one session under a metric definition.

Each Metric stores exactly one value per (metric, session) pair. Calling publish() a second time for the same metric name and session_id replaces the previous value (upsert semantics). This makes metrics suitable for recording per-session summary statistics that are computed once (or updated as reprocessing happens), not for streaming time-series data.

Recording a metric requires a MetricDefinition to already exist under the given name. If the definition doesn’t exist it will be created automatically.

Querying metrics (query()) returns the data points with a session timestamp in the given range. Aggregating metrics (aggregate()) groups sessions by the calendar period their stored timestamp falls into and applies a summary function (sum, mean, max, min, or count) across the values in each period.

Note

Metric instances should not be constructed directly. Obtain them via publish() or query().

Parameters:

• record (roboto.domain.metrics.record.MetricRecord)

• roboto_client (Optional[roboto.http.RobotoClient])

classmethod aggregate(name, period, aggregation, start_time, end_time, time_filter=MetricTimeFilter.EndTime, include_device_ids=NotSet, include_session_ids=NotSet, include_invocation_ids=NotSet, owner_org_id=None, roboto_client=None)

Aggregate a metric across sessions, grouped by calendar period.

Sessions whose session_min_timestamp_ns or session_max_timestamp_ns (selected via time_filter) falls inside the [start_time, end_time) window are grouped into UTC calendar buckets sized by period, and the chosen NumericAggregation is applied to the values in each bucket.

The server snaps the requested window outward to whole-period boundaries to guarantee apples-to-apples comparisons. All time period buckets always cover their complete calendar period. For example,

• a monthly aggregation requested between Jan 15 – Mar 15 will return aggregated data for all of

  January, February, and March.

• a quarterly aggregation from Apr 27 - Dec 28 will return aggregated data for all of Q2, Q3, and Q4.

Parameters:

• name (str) – Name of the metric definition to aggregate.

• period (roboto.domain.metrics.record.AggregationPeriod) – Calendar bucket size to group observations by.

• aggregation (roboto.domain.metrics.record.NumericAggregation) – Function to apply to values in each bucket.

• start_time (roboto.time.Time) – Inclusive start of the aggregation window. Accepts any Time value.

• end_time (roboto.time.Time) – Exclusive end of the aggregation window. Same input shape as start_time.

• time_filter (roboto.domain.metrics.record.MetricTimeFilter) – Whether to match the window against each session’s start time or end time. Defaults to end time.

• include_device_ids (Optional[Union[list[str], roboto.sentinels.NotSetType]]) – Restrict to specific device IDs, or None to match only rows with no device_id.

• include_session_ids (Union[list[str], roboto.sentinels.NotSetType]) – Restrict to specific session IDs.

• include_invocation_ids (Optional[Union[list[str], roboto.sentinels.NotSetType]]) – Restrict to specific invocation IDs, or None to match only rows with no invocation_id.

• owner_org_id (Optional[str]) – Organization that owns the metric data. Defaults to the authenticated caller’s organization.

• roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

One NumericAggregateMetricRecord per period bucket that contains at least one observation, sorted by start_time ascending.

Raises:

RobotoNotFoundException – No metric with this name exists in the organization.

Return type:

list[roboto.domain.metrics.record.NumericAggregateMetricRecord]

Examples

Daily max CPU usage over a month, passing datetime directly:

>>> import datetime
>>> from roboto.domain.metrics import (
...     AggregationPeriod,
...     Metric,
...     NumericAggregation,
... )
>>> for bucket in Metric.aggregate(
...     name="cpu.usage_max",
...     period=AggregationPeriod.Daily,
...     aggregation=NumericAggregation.Max,
...     start_time=datetime.datetime(2026, 5, 1, tzinfo=datetime.timezone.utc),
...     end_time=datetime.datetime(2026, 6, 1, tzinfo=datetime.timezone.utc),
... ):
...     print(bucket.start_time, bucket.value)

property device_id: str | None

Return type:

Optional[str]

classmethod get_by_session(session_id, owner_org_id=None, roboto_client=None)

Return every metric published to session_id.

Parameters:

• session_id (str) – Session whose metrics to fetch.

• owner_org_id (Optional[str]) – Organization that owns the session. Defaults to the authenticated caller’s organization.

• roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

One Metric per matching (metric_definition, session) pair. May be empty. Order is unspecified.

Return type:

list[Metric]

Examples

>>> from roboto.domain.metrics import Metric
>>> for m in Metric.get_by_session("ss_abc123"):
...     print(m.metric_id, m.value)

property invocation_id: str | None

Return type:

Optional[str]

property max_timestamp_ns: int | None

Return type:

Optional[int]

property metric_id: str

Return type:

str

property min_timestamp_ns: int | None

Return type:

Optional[int]

property name: str

Return type:

str

property org_id: str

Return type:

str

classmethod publish(session_id, metrics, device_id=NotSet, caller_org_id=None, roboto_client=None)

Record metric values for a session in a single network call.

Each (metric, session) pair is upserted: republishing under the same name and session_id replaces the previous value.

If a metric definition does not already exist for a given name it is created automatically. When called from within a Roboto action, successfully inserted records are automatically linked to the action invocation.

Parameters:

• session_id (str) – Session to attach every published value to.

• metrics (list[roboto.domain.metrics.record.MetricEntry]) – Metric names and values to record.

• device_id (Union[roboto.sentinels.NotSetType, Optional[str]]) – Device to associate with each published metric, or None to associate no device with the metric. When omitted, Roboto attempts to infer a device from the session’s attached devices. If the session has more than 1 device, device_id must be provided explicitly for each metric, or a RobotoInvalidRequestException: will be raised.

• caller_org_id (Optional[str]) – Organization context for the request. Defaults to the authenticated caller’s organization.

• roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

A BulkPublishMetricsResult with succeeded and failed lists. Items whose values are invalid, or whose names contain characters outside the URL-safe set, appear in failed; the remaining items are recorded and returned in succeeded.

Raises:

• RobotoNotFoundException – session_id does not exist in the caller’s organization.

• RobotoInvalidRequestException – device_id was omitted and the session has more than one attached devices.

Return type:

BulkPublishMetricsResult

Examples

Publish with an explicit device:

>>> from roboto.domain.metrics import Metric, MetricEntry
>>> result = Metric.publish(
...     session_id="ss_abc123",
...     metrics=[MetricEntry(name="cpu.usage_max", value=87.2)],
...     device_id="dv_robot01",
... )
>>> len(result.succeeded)
1

Let the server infer the device from the session’s single attached device:

>>> Metric.publish(
...     session_id="ss_abc123",
...     metrics=[MetricEntry(name="memory.peak_mb", value=2048.0)],
... )

Record values that are not tied to any device:

>>> Metric.publish(
...     session_id="ss_abc123",
...     metrics=[MetricEntry(name="run.duration_s", value=42.0)],
...     device_id=None,
... )

property published: datetime.datetime

Return type:

datetime.datetime

property published_by: str

Return type:

str

classmethod query(name, start_time=None, end_time=None, time_filter=MetricTimeFilter.EndTime, max_results=MAX_METRIC_LIST_RESULTS, include_device_ids=NotSet, include_session_ids=NotSet, include_invocation_ids=NotSet, owner_org_id=None, roboto_client=None)

Yield stored metric values whose session time falls in a range.

The time window is matched against either session_min_timestamp_ns or session_max_timestamp_ns on each metric row depending on time_filter.

This method auto-paginates: max_results is the page size (capped at MAX_METRIC_LIST_RESULTS), not a total result cap. The generator continues fetching pages until the server reports no more data.

Parameters:

• name (str) – Name of the metric definition to query.

• start_time (Optional[roboto.time.Time]) – Inclusive start of the query window. Accepts any Time value (int Unix-epoch nanoseconds, datetime, ISO 8601 string, decimal seconds, etc.). Defaults to None (the Unix epoch).

• end_time (Optional[roboto.time.Time]) – Exclusive end of the query window. Same input shape as start_time. Defaults to None (now).

• time_filter (roboto.domain.metrics.record.MetricTimeFilter) – Whether to match the window against the session’s start time or end time. Defaults to end time.

• max_results (int) – Page size — number of data points per HTTP request. Total results are unbounded; pagination is automatic.

• include_device_ids (Optional[Union[list[str], roboto.sentinels.NotSetType]]) – Restrict to specific device IDs, or None to match only rows with no device_id.

• include_session_ids (Union[list[str], roboto.sentinels.NotSetType]) – Restrict to specific session IDs.

• include_invocation_ids (Optional[Union[list[str], roboto.sentinels.NotSetType]]) – Restrict to specific invocation IDs, or None to match only rows with no invocation_id.

• owner_org_id (Optional[str]) – Organization that owns the metric data. Defaults to the authenticated caller’s organization.

• roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Yields:

One Metric per matching session, sorted by session time ascending with session_id as a deterministic tiebreaker.

Raises:

RobotoNotFoundException – No metric with this name exists in the organization.

Return type:

collections.abc.Generator[Metric, None, None]

Examples

Query a metric over a single day, passing datetime directly:

>>> import datetime
>>> from roboto.domain.metrics import Metric
>>> for m in Metric.query(
...     name="cpu.usage_max",
...     start_time=datetime.datetime(2026, 5, 1, tzinfo=datetime.timezone.utc),
...     end_time=datetime.datetime(2026, 5, 2, tzinfo=datetime.timezone.utc),
... ):
...     print(m.session_id, m.value)

Or with an ISO 8601 string:

>>> all_records = list(
...     Metric.query(
...         name="cpu.usage_max",
...         start_time="2026-05-01T00:00:00Z",
...         end_time="2026-05-02T00:00:00Z",
...     )
... )

property record: roboto.domain.metrics.record.MetricRecord

Return type:

roboto.domain.metrics.record.MetricRecord

property session_id: str

Return type:

str

property value: float

Return type:

float

class roboto.domain.metrics.metric.MetricDefinition(record, roboto_client)

A named schema for a metric tracked across sessions and devices.

Metric definitions are org-scoped schemas that describe a single measurable quantity. They act as the registry entry that all Metric data points reference. Every metric definition has a unique name within an organization, and an optional human-readable description.

Metric definitions are created once per org and reused across many sessions. Use create() to register a definition the first time, and update() to change its description later. for_org() lists all definitions that belong to an organization.

Note

MetricDefinition instances should not be constructed directly. Always obtain them via create(), get(), or for_org().

Parameters:

• record (roboto.domain.metrics.record.MetricDefinitionRecord)

• roboto_client (Optional[roboto.http.RobotoClient])

classmethod create(name, description=None, caller_org_id=None, roboto_client=None)

Create a new metric definition in the caller’s organization.

Parameters:

• name (str) – Unique metric name. Must contain only URL-safe characters (A–Z, a–z, 0–9, -, ., _, ~). Dots are conventional namespace separators, e.g. cpu.usage_pct.

• description (Optional[str]) – Optional human-readable description of what the metric measures and its units.

• caller_org_id (Optional[str]) – Organization to create the definition in. Defaults to the authenticated caller’s organization.

• roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

The newly created MetricDefinition.

Raises:

RobotoConflictException – A definition with this name already exists in the organization.

Return type:

MetricDefinition

Examples

>>> MetricDefinition.create(
...     name="cpu.usage_max",
...     description="Peak CPU usage percentage recorded during the session.",
... )

delete()

Delete this metric definition and all of its associated data points.

Warning

This operation is irreversible. All Metric data points recorded under this name will be permanently removed.

Examples

>>> definition = MetricDefinition.get("cpu.usage_max")
>>> definition.delete()

Return type:

None

property description: str | None

Return type:

Optional[str]

classmethod for_org(owner_org_id, roboto_client=None)

Yield all metric definitions belonging to an organization.

Parameters:

• owner_org_id (str) – Organization that owns the metric definitions to enumerate.

• roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Yields:

Each MetricDefinition belonging to owner_org_id.

Return type:

collections.abc.Generator[MetricDefinition, None, None]

Examples

>>> for definition in MetricDefinition.for_org("og_myorg"):
...     print(definition.name, "-", definition.description)

classmethod get(name, owner_org_id=None, roboto_client=None)

Retrieve an existing metric definition by name.

Parameters:

• name (str) – Name of the metric definition to retrieve. Must match exactly (case-sensitive) the name used when the definition was created.

• owner_org_id (Optional[str]) – Organization that owns the definition. Defaults to the authenticated caller’s organization.

• roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client to use. Defaults to the client configured in the environment.

Returns:

The MetricDefinition with the given name.

Raises:

RobotoNotFoundException – No definition with this name exists in the organization.

Return type:

MetricDefinition

Examples

>>> definition = MetricDefinition.get("cpu.usage_max")

property metric_id: str

Return type:

str

property name: str

Return type:

str

property org_id: str

Return type:

str

update(description=NotSet)

Update the description of this definition.

Parameters:

description (Optional[Union[roboto.sentinels.NotSetType, str]]) – New human-readable description, None to clear, or NotSet to leave unchanged.

Return type:

None

Examples

>>> definition = MetricDefinition.get("cpu.usage_max")
>>> definition.update(description="Peak CPU usage percentage recorded during the session.")

roboto.domain.metrics.metric.logger