roboto.domain.topics.topic_schema#

Module Contents#

class roboto.domain.topics.topic_schema.TopicSchema(record, fields, roboto_client)#

Describes the field structure of a topic’s messages.

A topic schema is identified by a name (e.g., "sensor_msgs/Imu") and a content-based checksum deterministically derived from its fields. Schemas are deduplicated within an organization: topics whose fields share the same names, paths, and data types reference the same schema.

Use from_id() when you already know the schema_id, or for_topic() to retrieve the schema associated with a specific topic. Topic.get_schema() is also a convenient entry point.

Examples

Retrieve a topic’s schema and inspect its fields:

>>> from roboto.domain.topics import TopicSchema
>>> schema = TopicSchema.for_topic(topic_id="tp_abc123")
>>> print(schema.name, schema.checksum)
>>> for field in schema.fields:
...     print(field.path_in_schema, field.data_type)
Parameters:
property checksum: str#

Content-based checksum of the schema’s field set.

Return type:

str

property fields: list[roboto.domain.topics.record.SchemaFieldRecord]#

Field definitions belonging to this schema.

Return type:

list[roboto.domain.topics.record.SchemaFieldRecord]

classmethod for_topic(topic_id, owner_org_id=None, roboto_client=None)#

Retrieve the schema associated with a topic.

Parameters:
  • topic_id (str) – Unique identifier of the topic whose schema to retrieve.

  • owner_org_id (Optional[str]) – Organization that owns the topic. Required for cross-org access.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

A TopicSchema for the topic.

Raises:

RobotoNotFoundException – The topic does not exist, or has no schema associated with it.

Return type:

TopicSchema

classmethod from_id(schema_id, owner_org_id=None, roboto_client=None)#

Retrieve a schema by its ID.

Parameters:
  • schema_id (str) – Unique identifier of the schema to retrieve.

  • owner_org_id (Optional[str]) – Organization that owns the schema. Required when the caller belongs to multiple orgs.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

A TopicSchema for the given schema_id.

Raises:

RobotoNotFoundException – No schema with this ID exists in the scoped org.

Return type:

TopicSchema

Examples

>>> from roboto.domain.topics import TopicSchema
>>> schema = TopicSchema.from_id("ts_abc123")
>>> for field in schema.fields:
...     print(field.path_in_schema, field.data_type)
property name: str | None#

Informational label for the schema (e.g. "sensor_msgs/Imu"). Not part of identity; may be None.

Return type:

Optional[str]

property record: roboto.domain.topics.record.TopicSchemaRecord#

Underlying schema record.

Return type:

roboto.domain.topics.record.TopicSchemaRecord

property schema_id: str#

Unique identifier for this schema.

Return type:

str