Python SDK Reference¶
This section is generated from the installed zelos_sdk package.
Use the sidebar to browse modules, or jump directly:
Package Overview¶
zelos_sdk
¶
TraceEventFieldMetadata
¶
Metadata describing a field in a trace event schema.
This class defines the structure of a field within an event schema, including its name, data type, and optional unit of measurement.
Args: name (str): The field name. data_type (DataType): The data type for the field. unit (Optional[str]): Optional unit of measurement.
Examples: >>> # Define a field for HTTP status code >>> status_field = TraceEventFieldMetadata("status_code", DataType.Int32) >>> >>> # Define a field with a unit of measurement >>> duration_field = TraceEventFieldMetadata( ... "duration_ms", DataType.Float64, "milliseconds")
TraceSender
¶
Communication channel for sending trace events.
This class is typically obtained from a TracePublishClient and passed to a TraceSource during creation. It handles the underlying message transport.
Note: Users generally don't need to interact with this class directly; it's used internally to connect TraceSource to TracePublishClient.
TraceSourceCacheLast(name)
¶
A TraceSource wrapper that caches the last value of each field.
Example: source = TraceSourceCacheLast("motor_controller") source.add_event("motor_stats", [ TraceEventFieldMetadata("rpm", DataType.Float64), TraceEventFieldMetadata("torque", DataType.Float64, "Nm") ])
# Log some data
source.log("motor_stats", {"rpm": 3500.0, "torque": 42.8})
# Access cached values
assert source.motor_stats.rpm.get() == 3500.0
assert source.motor_stats.torque.get() == 42.8
# Dictionary-style access
assert source["motor_stats"].rpm.get() == 3500.0
assert source["motor_stats/rpm"] == 3500.0
# Log via event object
source.motor_stats.log(rpm=3250.0, torque=45.2)
__getattr__(name)
¶
Get an event by attribute access. Only returns existing events.
__getitem__(key)
¶
Support dictionary-style access for events and fields.
Examples: source["my_event"] # Returns TraceSourceCacheLastEvent source["my_event/subevent"] # Returns nested TraceSourceCacheLastEvent source["my_event/field"] # Returns TraceSourceCacheLastField object source["event/sub/field"] # Returns deeply nested TraceSourceCacheLastField object
add_event(name, schema, conditions=None)
¶
Add an event to the source and create a cached version.
get_source()
¶
Get the underlying TraceSource.
log(name, data)
¶
Log data to an event and update the cache.
log_at(time_ns, name, data)
¶
Log data to an event at a specific time and update the cache.
set_default_log_condition(condition=DefaultLogCondition())
¶
Set a default log condition that applies to all fields without explicit conditions.
This will also update existing fields that currently use the default condition to use the new default.
Args: condition: The default logging condition to use, or None to disable default conditions.
TraceSourceCacheLastEvent(name, event, source, conditions=None)
¶
A cached event that provides access to fields and submessages.
Example: event = source.motor_stats event.rpm.get() # Get field value event.thermal.temp.get() # Get nested field value event.log(rpm=3500) # Log new values
TraceSourceCacheLastField(name, metadata, full_path, condition=None, uses_default=False)
¶
A cached field that stores the last logged value.
Example: field = event.rpm # Get field field.get() # Get cached value field.name # Get full path like "motor_stats.rpm"
TraceStdout
¶
Python wrapper for the stdout trace sink.
This sink outputs trace events to stdout with configurable log levels. It subscribes to all trace events from the router and formats them as structured log messages.
The sink uses context management and should be used with a with statement
to ensure proper resource cleanup and automatic start/stop of trace capture.
Examples: >>> # Basic usage with default settings (info level) >>> with TraceStdout() as sink: ... # Trace events will be logged to stdout ... pass >>> >>> # Custom log level and batch configuration >>> with TraceStdout(log_level="debug", batch_size=500, batch_timeout_ms=2000) as sink: ... # Trace events will be logged with custom settings ... pass
__repr__()
¶
String representation of the sink.
close()
¶
Stop the stdout sink and finalize trace capture.
This method gracefully shuts down the sink and cancels background tasks. It's automatically called when exiting the context manager.
Returns: None
open()
¶
Start the stdout sink and begin capturing events.
This method subscribes to the trace router and starts a background task to process and output trace events to stdout. It's automatically called when entering the context manager (with statement).
Returns: None
Raises: RuntimeError: If the sink cannot be initialized.
TraceWriter
¶
Python wrapper for the TraceWriter.
This writer manages writing trace events to a local file, with support for batching and buffering. It can be used with a TraceSource to capture events for later analysis.
The writer uses context management and should be used with a with statement
to ensure proper resource cleanup and automatic start/stop of trace capture.
Examples: >>> # Basic usage with default settings >>> with TraceWriter("my_trace.trz") as writer: ... # Trace events will be captured automatically ... pass >>> >>> # Custom batch configuration >>> with TraceWriter("my_trace.trz", batch_size=500, batch_timeout_ms=2000) as writer: ... # Trace events will be captured with custom batch settings ... pass
__repr__()
¶
String representation of the writer.
close()
¶
Stop the trace writer and finalize trace capture.
This method gracefully shuts down the writer, cancels background tasks, and ensures all buffered events are written to the trace file. It's automatically called when exiting the context manager.
Returns: None
Note: This method is called automatically by exit when using the context manager pattern.
open()
¶
Start the trace writer and begin capturing events.
This method initializes the writer and starts background tasks for batching and writing trace events. It's automatically called when entering the context manager (with statement).
Returns: None
Raises: RuntimeError: If the writer cannot be initialized.
Note: This method is called automatically by enter when using the context manager pattern.
get_global_router_sender()
¶
Get the global default trace router sender
Returns: TraceSender: The global router's sender
Examples: >>> sender = get_global_router_sender()
init(name=None, *, url=None, client_config=None, log_level=None, trace=True, actions=False, block=False)
¶
Initialize the Zelos SDK tracing and actions systems.
Args: name: A unique identifier for your application. Defaults to "python". client_config: Configuration options for the TracePublishClient. Can include: url, batch_size, batch_timeout_ms. log_level: Logging level to enable, None disables logging. trace: Whether to initialize the trace system. Defaults to True. actions: Whether to initialize the actions system. Defaults to False. block: Whether to block the current thread until exit. Defaults to False.
Examples: >>> # Initialize with defaults >>> init() >>> >>> # Initialize with custom name >>> init("my_app") >>> >>> # Initialize with custom config >>> init( ... "my_app", ... client_config=TracePublishClientConfig( ... url="grpc://localhost:2300", ... ), ... log_level="debug" ... ) >>> >>> # Initialize only logging, no tracing or actions >>> init(log_level="debug", trace=False) >>> >>> # Initialize actions and block until exit >>> init(actions=True, block=True)