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.

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#