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.

condition: roboto.query.ConditionType | None = None#

Condition, or nested group of conditions, narrowing which data points are aggregated.

Applied to individual data points rather than to bucket results, so it changes each bucket’s value and total. A period whose data points are all filtered out yields no bucket at all, so a filtered aggregation can return fewer buckets than an unfiltered one over the same window. See condition for the accepted fields and the treatment of data points published without a device.

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.

name: str#

Unique metric name.

unit: MetricUnit | None = None#

Unit of measure for values recorded under this metric, e.g. "%", "ms". Capped at 63 characters. None means unitless.

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, condition=None, 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.

  • condition (Optional[roboto.query.ConditionType]) – Restrict the aggregated data points to those whose session, producing device, or session’s collections match this Condition or ConditionGroup. It narrows what each bucket aggregates without moving the window’s period boundaries; a bucket left with no matching data points is omitted from the result. See condition for the accepted fields, and for how collection conditions evaluate when a session belongs to several collections or to none.

  • 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.

  • RobotoIllegalArgumentException – condition references a field or comparator the request does not accept, or a custom field that is either undefined in your organization or defined but not in the Ready state.

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)

The same aggregation over data points published from a device in the delivery fleet. The Device custom field fleet must be defined by your organization and moved to Ready; the aggregation raises if it has not been. A data point published without a device carries no fleet, so Equals drops it, and NotEquals drops it too: only IsNull and NotExists match a data point with no device. See condition for the full field and comparator rules:

>>> from roboto.query import Comparator, Condition
>>> delivery_buckets = Metric.aggregate(
...     name="cpu.usage_max",
...     period=AggregationPeriod.Daily,
...     aggregation=NumericAggregation.Max,
...     start_time="2026-05-01T00:00:00Z",
...     end_time="2026-06-01T00:00:00Z",
...     condition=Condition(
...         field="device.custom.fleet",
...         comparator=Comparator.Equals,
...         value="delivery",
...     ),
... )
>>> for bucket in delivery_buckets:
...     print(bucket.start_time, bucket.end_time, bucket.value, bucket.total)
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, descending=False, include_device_ids=NotSet, include_session_ids=NotSet, include_invocation_ids=NotSet, condition=None, 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.

  • descending (bool) – Yield the most recent sessions first instead of the oldest first. Applies across the whole result set, not just within a page.

  • 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.

  • condition (Optional[roboto.query.ConditionType]) – Restrict to data points whose session, producing device, or session’s collections match this Condition or ConditionGroup. See condition for the accepted fields, and for how collection conditions evaluate when a session belongs to several collections or to none.

  • 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 by default, descending when descending is set — with session_id as a deterministic tiebreaker.

Raises:

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

  • RobotoIllegalArgumentException – condition references a field or comparator the request does not accept, or a custom field that is either undefined in your organization or defined but not in the Ready state.

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",
...     )
... )

Take just the 10 most recent sessions:

>>> import itertools
>>> recent = list(itertools.islice(Metric.query(name="cpu.usage_max", descending=True), 10))

Restrict to data points from production-tagged sessions that either ran in the EMEA region or were produced by a device at the Berlin site. Both custom fields, region on Sessions and site on Devices, must be defined by your organization and moved to Ready; the query raises if either has not been:

>>> from roboto.query import Comparator, Condition, ConditionGroup, ConditionOperator
>>> for m in Metric.query(
...     name="cpu.usage_max",
...     condition=ConditionGroup(
...         operator=ConditionOperator.And,
...         conditions=[
...             Condition(
...                 field="session.tags",
...                 comparator=Comparator.Contains,
...                 value="production",
...             ),
...             ConditionGroup(
...                 operator=ConditionOperator.Or,
...                 conditions=[
...                     Condition(
...                         field="session.custom.region",
...                         comparator=Comparator.Equals,
...                         value="emea",
...                     ),
...                     Condition(
...                         field="device.custom.site",
...                         comparator=Comparator.Equals,
...                         value="berlin",
...                     ),
...                 ],
...             ),
...         ],
...     ),
... ):
...     print(m.session_id, m.value)
property record: roboto.domain.metrics.record.MetricRecord#
Return type:

roboto.domain.metrics.record.MetricRecord

property session_id: str#
Return type:

str

property unit: str | None#
Return type:

Optional[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, unit=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.

  • unit (Optional[str]) – Optional unit of measure for values recorded under this metric, e.g. "%", "ms", "m/s". Free-form and unvalidated. Omit for a unitless metric.

  • 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 recorded during the session.",
...     unit="%",
... )
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

property unit: str | None#
Return type:

Optional[str]

update(description=NotSet, unit=NotSet)#

Update the mutable attributes of this definition.

Parameters:
Return type:

None

Examples

>>> definition = MetricDefinition.get("cpu.usage_max")
>>> definition.update(description="Peak CPU usage recorded during the session.", unit="%")
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.

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.

unit: str | None = None#

Unit of measure for every value recorded under this metric, e.g. "%", "ms", "m/s". Free-form and unvalidated; None means unitless.

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.

unit: str | None = None#

Unit of measure for value. Resolved server-side from the parent MetricDefinitionRecord, like name, so callers can label a value without a second lookup. None means unitless.

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.

unit: str | None = None#

Unit of measure for value, resolved from the aggregated metric’s definition alongside name. None means unitless.

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)

condition: roboto.query.ConditionType | None = None#

Condition, or nested group of conditions, narrowing which data points are returned.

Every field must be prefixed with the entity it filters on, singular or plural; a bare field name such as name is rejected. The available fields are:

  • session.<field>: session_id (alias id), name, min_timestamp_ns (alias start_time), max_timestamp_ns (alias end_time), duration, created, created_by, modified, modified_by, tags. The two timestamp bounds accept anything to_epoch_nanoseconds() converts; duration takes an integer count of nanoseconds.

  • device.<field>: device_id (alias id), tags, created, created_by, modified, modified_by, metadata (including dotted paths beneath it). Reads the device that published the data point, not the devices attached to its session.

  • session.custom.<name> / device.custom.<name> / collection.custom.<name>: a custom field in the Ready state.

  • collection.collection_id (alias collection.id): the data point’s session belongs to that collection. Equals and NotEquals only.

A session belongs to any number of collections, so a collection.* condition quantifies over that set: a data point matches when its session belongs to at least one collection satisfying the condition. A negated comparator (NotEquals, NotContains, NotLike) means the session belongs to no collection satisfying the positive form, so a session in no collection at all matches every negated collection condition. IsNull and NotExists likewise mean no collection the session belongs to carries a value for the field.

A data point published without a device matches a device.* condition only under IsNull and NotExists. Every other comparator asks what the device’s field holds, NotEquals, NotContains, and NotLike included, so a data point with no device, or a device carrying no value for the field, is excluded. Write device.<field> NotEquals x OR device.<field> IsNull to match both.

Any other field, a comparator the field’s type does not accept, a value the field cannot convert, or a Not group raises RobotoIllegalArgumentException.

descending: bool = False#

Return the most recent data points first, instead of the oldest first.

Ordering is by the column selected by time_filter, with session_id as a deterministic tiebreaker. A page token is bound to the direction that issued it: replaying one with a different descending value is rejected with RobotoInvalidRequestException, because the keyset cursor is only meaningful in the direction it was emitted from.

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].

unit: roboto.sentinels.NotSetType | MetricUnit | None#

New unit of measure (max 63 characters), None to clear, or NotSet to leave unchanged.

On this page