roboto.domain.metrics.record#

Module Contents#

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

Bases: pydantic.BaseModel

!!! abstract “Usage Documentation”: [Models](../concepts/models.md)

A base class for creating Pydantic models.

Parameters:: data (Any)

__class_vars__#: The names of the class variables defined on the model.

__private_attributes__#: Metadata about the private attributes of the model.

__signature__#: The synthesized __init__ [Signature][inspect.Signature] of the model.

__pydantic_complete__#: Whether model building is completed, or if there are still undefined fields.

__pydantic_core_schema__#: The core schema of the model.

__pydantic_custom_init__#: Whether the model has a custom __init__ function.

__pydantic_decorators__#: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.

__pydantic_generic_metadata__#: A dictionary containing metadata about generic Pydantic models. The origin and args items map to the [__origin__][genericalias.__origin__] and [__args__][genericalias.__args__] attributes of [generic aliases][types-genericalias], and the parameter item maps to the __parameter__ attribute of generic classes.

__pydantic_parent_namespace__#: Parent namespace of the model, used for automatic rebuilding of models.

__pydantic_post_init__#: The name of the post-init method for the model, if defined.

__pydantic_root_model__#: Whether the model is a [RootModel][pydantic.root_model.RootModel].

__pydantic_serializer__#: The pydantic-core SchemaSerializer used to dump instances of the model.

__pydantic_validator__#: The pydantic-core SchemaValidator used to validate instances of the model.

__pydantic_fields__#: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.

__pydantic_computed_fields__#: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.

__pydantic_extra__#: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to ‘allow’.

__pydantic_fields_set__#: The names of fields explicitly set during instantiation.

__pydantic_private__#: Values of private attributes set on the model instance.

end_time: int#: Exclusive end of this period bucket, in Unix-epoch nanoseconds (UTC).

metric_id: str#: Identifier of the aggregated metric definition.

name: str#: Name of the aggregated metric.

period: AggregationPeriod#: Calendar bucket size used for this aggregation.

start_time: int#: Inclusive start of this period bucket, in Unix-epoch nanoseconds (UTC).

total: int#: Number of raw observations that contributed to this bucket.

class roboto.domain.metrics.record.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.record.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.record.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.record.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.record.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.record.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.record.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.record.MetricTimeFilter#

Bases: roboto.compat.StrEnum

Enum where members are also (and must be) strings

EndTime = 'end_time'#

StartTime = 'start_time'#

class roboto.domain.metrics.record.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.record.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.record.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.record.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.record.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.record.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.record.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.record.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].