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 theschema_id, orfor_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:
fields (list[roboto.domain.topics.record.SchemaFieldRecord])
roboto_client (roboto.http.RobotoClient)
- 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:
- 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
TopicSchemafor the topic.- Raises:
RobotoNotFoundException – The topic does not exist, or has no schema associated with it.
- Return type:
- 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
TopicSchemafor the givenschema_id.- Raises:
RobotoNotFoundException – No schema with this ID exists in the scoped org.
- Return type:
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 beNone.- Return type:
Optional[str]
- property record: roboto.domain.topics.record.TopicSchemaRecord#
Underlying schema record.
- Return type:
- property schema_id: str#
Unique identifier for this schema.
- Return type:
str