roboto.domain.metrics#

Submodules#

Package Contents#

class roboto.domain.metrics.AggregateMetricsRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for a numeric metric aggregation.

Parameters:

data (Any)

aggregation: NumericAggregation#

Aggregation function to apply to the values in each bucket.

end_time_ns: int#

Exclusive end of the aggregation window, in Unix-epoch nanoseconds (UTC). Built from aggregate()’s end_time parameter the same way.

include_device_ids: list[str] | roboto.sentinels.NotSetType | None#

Filter to observations from specific device IDs, None for null device_id only.

include_invocation_ids: list[str] | roboto.sentinels.NotSetType | None#

Filter to observations from specific invocation IDs, None for null invocation_id only.

include_session_ids: list[str] | roboto.sentinels.NotSetType#

Filter to observations for specific session IDs. None is not a valid value: metrics.session_id is non-nullable, so there is no “null session” subset to filter on. Omit (leave as NotSet) for no filter, or pass a list of IDs.

model_config#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

Name of the metric to aggregate.

period: AggregationPeriod#

Calendar bucket size to group observations by.

start_time_ns: int#

Inclusive start of the aggregation window, in Unix-epoch nanoseconds (UTC). Built by aggregate() from its start_time parameter via to_epoch_nanoseconds().

time_filter: MetricTimeFilter#

Whether to filter by session start time or end time.

class roboto.domain.metrics.AggregationPeriod#

Bases: roboto.compat.StrEnum

Calendar bucket size used when grouping metric observations.

All aggregation start/end times are based on UTC time.

Daily = 'daily'#

One bucket per calendar day.

Monthly = 'monthly'#

One bucket per calendar month.

Quarterly = 'quarterly'#

One bucket per calendar quarter (three months).

Weekly = 'weekly'#

One bucket per calendar week.

Yearly = 'yearly'#

One bucket per calendar year.

class roboto.domain.metrics.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.CreateMetricDefinitionRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to create a metric definition.

Parameters:

data (Any)

description: str | None = None#

Human-readable description of what the metric measures and its units.

name: str#

Unique metric name.

roboto.domain.metrics.MAX_METRIC_LIST_RESULTS: int = 10000#

Upper bound on the page size accepted by metric query and list calls.

query() auto-paginates with this value as the default page size, so total result-set size is unbounded. Callers can request smaller pages by setting max_results.

get_by_session() does not paginate and is still capped at this many rows; sessions with more data points should use the paginated query() instead.

class roboto.domain.metrics.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:
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:
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.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:
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.")
class roboto.domain.metrics.MetricDefinitionRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of a metric definition.

Parameters:

data (Any)

created: datetime.datetime#

Timestamp when this metric definition was created.

created_by: str#

User or service account that created this metric definition.

description: str | None = None#

Human-readable description of what the metric measures and its units.

metric_id: str#

Unique identifier for this metric definition.

modified: datetime.datetime#

Timestamp when this metric definition was last modified.

modified_by: str#

User or service account that last modified this metric definition.

name: str#

Unique name for this metric.

org_id: str#

Organization that owns this metric definition.

class roboto.domain.metrics.MetricEntry(/, **data)#

Bases: pydantic.BaseModel

A single name+value pair within a bulk metric publish.

Parameters:

data (Any)

name: str#

Name of the metric definition to record a value for. If the definition does not exist, it is auto-created.

value: float#

Observed numeric value.

class roboto.domain.metrics.MetricRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of a metric data point.

Parameters:

data (Any)

device_id: str | None = None#

Device that produced the data.

invocation_id: str | None = None#

Action invocation that produced this data point, if any.

max_timestamp_ns: int | None = None#

Upper bound of the source session’s aggregate timestamps, in Unix-epoch nanoseconds. None until the session has at least one file contribution. Mirrors max_timestamp_ns.

metric_id: str#

Identifier of the metric definition this data point belongs to.

min_timestamp_ns: int | None = None#

Lower bound of the source session’s aggregate timestamps, in Unix-epoch nanoseconds. None until the session has at least one file contribution. Mirrors min_timestamp_ns.

name: str#

Human-readable name of the metric definition this data point belongs to. Resolved server-side from the parent MetricDefinitionRecord so callers do not need a second lookup to display the metric name alongside the value.

org_id: str#

Organization that owns this metric data point.

published: datetime.datetime#

Timestamp when this data point was published to the platform.

published_by: str#

User or service account that published this data point.

session_id: str#

Session this metric is associated with.

value: float#

Observed numeric value.

class roboto.domain.metrics.MetricTimeFilter#

Bases: roboto.compat.StrEnum

Enum where members are also (and must be) strings

EndTime = 'end_time'#
StartTime = 'start_time'#
class roboto.domain.metrics.NumericAggregateMetricRecord(/, **data)#

Bases: AggregateMetricRecord

A wire-transmissible representation of one period bucket in a numeric metric aggregation.

Parameters:

data (Any)

aggregation: NumericAggregation#

Aggregation function that was applied to produce this record.

value: float#

Aggregated result for this bucket.

class roboto.domain.metrics.NumericAggregateMetricsResponse(/, **data)#

Bases: pydantic.BaseModel

Response payload for a numeric metric aggregation request.

Parameters:

data (Any)

aggregation: NumericAggregation#

Aggregation function that was applied.

records: list[NumericAggregateMetricRecord]#

Period buckets returned by the aggregation, sorted by start_time ascending.

class roboto.domain.metrics.NumericAggregation#

Bases: roboto.compat.StrEnum

Aggregation function applied to numeric metric values within each period bucket.

Count = 'count'#

Count of observations in the bucket.

Max = 'max'#

Maximum value observed in the bucket.

Mean = 'mean'#

Arithmetic mean of all values in the bucket.

Min = 'min'#

Minimum value observed in the bucket.

Sum = 'sum'#

Sum of all values in the bucket.

class roboto.domain.metrics.PublishMetricsError(/, **data)#

Bases: pydantic.BaseModel

One failed item from a bulk metric publish.

Parameters:

data (Any)

error: str#

Human-readable description of why the insert failed.

name: str#

Name of the metric that failed to insert.

class roboto.domain.metrics.PublishMetricsRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to insert multiple metric data points in a single call.

Parameters:

data (Any)

device_id: roboto.sentinels.NotSetType | str | None#

Device that produced the data. When absent (NotSet), the server infers the device from the session’s attached devices: the request succeeds if exactly one device is attached and is rejected otherwise. Pass an explicit device ID or None to skip inference.

metrics: list[MetricEntry]#

Metric data points to insert.

model_config#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

session_id: str#

Session all metrics in this batch will be attached to.

class roboto.domain.metrics.PublishMetricsResponse(/, **data)#

Bases: pydantic.BaseModel

Server response from a bulk metric publish.

May contain a mix of successes and per-item failures if some metric values are invalid.

Parameters:

data (Any)

failed: list[PublishMetricsError]#
succeeded: list[MetricRecord]#
class roboto.domain.metrics.QueryMetricsRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to query raw metric data points.

Parameters:

data (Any)

end_time_ns: int | None = None#

Exclusive end of the query window, in Unix-epoch nanoseconds (UTC). Built from query()’s end_time parameter the same way. Defaults to None (now).

include_device_ids: list[str] | roboto.sentinels.NotSetType | None#

Filter to observations from specific device IDs, None for null device_id only.

include_invocation_ids: list[str] | roboto.sentinels.NotSetType | None#

Filter to observations from specific invocation IDs, None for null invocation_id only.

include_session_ids: list[str] | roboto.sentinels.NotSetType#

Filter to observations for specific session IDs. None is not a valid value: metrics.session_id is non-nullable, so there is no “null session” subset to filter on. Omit (leave as NotSet) for no filter, or pass a list of IDs.

max_results: int = None#

Maximum number of data points to return. Must be between 1 and MAX_METRIC_LIST_RESULTS (10,000).

model_config#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

Name of the metric to query.

start_time_ns: int | None = None#

Inclusive start of the query window, in Unix-epoch nanoseconds (UTC). Built by query() from its start_time parameter via to_epoch_nanoseconds(). Defaults to None (the Unix epoch).

time_filter: MetricTimeFilter#

Whether to filter by session start time or end time.

class roboto.domain.metrics.UpdateMetricDefinitionRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to update a metric definition.

Parameters:

data (Any)

description: roboto.sentinels.NotSetType | str | None#

New description, None to clear, or NotSet to leave unchanged.

model_config#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

On this page