roboto.domain.actions.invocation_record

Module Contents

ActionProvenance

class roboto.domain.actions.invocation_record.ActionProvenance(/, **data)#

Bases: pydantic.BaseModel

Provenance information for an action

Parameters

data Any

Attributes

ActionProvenance.digest

digest str | None = None #

ActionProvenance.name

name str #

ActionProvenance.org_id

org_id str #

DataSelector

class roboto.domain.actions.invocation_record.DataSelector(/, **data)#

Bases: pydantic.BaseModel

Selector for inputs (e.g. files) to an action invocation.

Parameters

data Any

Attributes

DataSelector.dataset_id

dataset_id str | None = None #

Dataset ID, needed for backward compatibility purposes. Prefer RoboQL: dataset_id = <the ID in double quotes>

DataSelector.ensure_not_empty()

ensure_not_empty()#

Return type

DataSelector

Attributes

DataSelector.ids

ids list[str] | None = None #

Specific input IDs (e.g. a dataset ID).

DataSelector.names

names list[str] | None = None #

Specific input names (e.g. topic names).

DataSelector.query

query str | None = None #

RoboQL query representing the desired inputs.

ExecutableProvenance

class roboto.domain.actions.invocation_record.ExecutableProvenance(/, **data)#

Bases: pydantic.BaseModel

Provenance information for an action executable

Parameters

data Any

Attributes

ExecutableProvenance.container_image_digest

container_image_digest str | None = None #

ExecutableProvenance.container_image_uri

container_image_uri str | None = None #

FileSelector

class roboto.domain.actions.invocation_record.FileSelector(/, **data)#

Bases: DataSelector

Selector for file inputs to an action invocation.

This selector type exists for backward compatibility purposes. We encourage you to use the query field to scope your input query to any dataset and/or file paths.

Parameters

data Any

FileSelector.ensure_not_empty()

ensure_not_empty()#

Return type

FileSelector

Attributes

FileSelector.paths

paths list[str] | None = None #

File paths or patterns. Prefer RoboQL: path LIKE <path pattern in double quotes>

InvocationDataSource

class roboto.domain.actions.invocation_record.InvocationDataSource(/, **data)#

Bases: pydantic.BaseModel

Abstracted data source that can be provided to an invocation.

Represents a source of input data for action invocations. The data source type determines how the ID should be interpreted (e.g., as a dataset ID).

Parameters

data Any

Attributes

InvocationDataSource.data_source_id

data_source_id str #

The ID of the data source. For Dataset type, this is a dataset ID.

InvocationDataSource.data_source_type

data_source_type InvocationDataSourceType #

The type of data source (currently only Dataset).

InvocationDataSource.is_unspecified()

is_unspecified()#

Check if this data source is unspecified.

Returns

bool

True if this is an unspecified data source, False otherwise.

InvocationDataSource.unspecified()

static unspecified()#

Returns a special value indicating that no invocation source is specified.

Returns

InvocationDataSource

An InvocationDataSource instance representing an unspecified data source.

InvocationDataSourceType

class roboto.domain.actions.invocation_record.InvocationDataSourceType(*args, **kwds)#

Bases: enum.Enum

Source of data for an action’s input binding.

Defines the type of data source that provides input data to an action invocation. Currently supports datasets, with potential for future expansion to other data source types.

Attributes

InvocationDataSourceType.Dataset

Dataset = 'Dataset' #

InvocationInput

class roboto.domain.actions.invocation_record.InvocationInput(/, **data)#

Bases: pydantic.BaseModel

Input specification for an action invocation.

An invocation may require no inputs at all, or some combination of Roboto files, topics, events, etc. Those are specified using selectors, which tell the invocation how to locate inputs. Selector choices include RoboQL queries (for maximum flexibility), as well as unique IDs or friendly names.

Note: support for certain input types is a work in progress, and will be offered in future Roboto platform releases.

At least one data selector must be provided in order to construct a valid InvocationInput instance.

Parameters

data Any

InvocationInput.ensure_not_empty()

ensure_not_empty()#

Return type

InvocationInput

Properties

InvocationInput.file_paths

file_paths list[str] #

Return type: list[str]

InvocationInput.file_query()

classmethod file_query(roboql_query)#

Specify file inputs using a RoboQL query.

Parameters

roboql_query str

Return type

InvocationInput

Attributes

InvocationInput.files

files FileSelector | list[FileSelector] | None = None #

File selectors.

InvocationInput.from_dataset_file_paths()

classmethod from_dataset_file_paths(dataset_id, file_paths)#

Specify file input by dataset ID and dataset-relative file paths or globs.

Parameters

dataset_id str

file_paths list[str]

Return type

InvocationInput

InvocationInput.from_session_id()

classmethod from_session_id(session_id)#

Specify session input by session ID.

Parameters

session_id str

Return type

InvocationInput

Properties

InvocationInput.safe_files

safe_files list[FileSelector] #

File selectors as a list. Empty if no such selectors are specified.

Return type: list[FileSelector]

InvocationInput.safe_sessions

safe_sessions list[DataSelector] #

Session selectors as a list. Empty if no such selectors are specified.

Return type: list[DataSelector]

InvocationInput.safe_topics

safe_topics list[DataSelector] #

Topic selectors as a list. Empty if no such selectors are specified.

Return type: list[DataSelector]

InvocationInput.session_query()

classmethod session_query(roboql_query)#

Specify session inputs using a RoboQL query.

Parameters

roboql_query str

Return type

InvocationInput

Attributes

InvocationInput.sessions

sessions DataSelector | list[DataSelector] | None = None #

Session selectors.

InvocationInput.topic_query()

classmethod topic_query(roboql_query)#

Specify topic inputs using a RoboQL query.

Parameters

roboql_query str

Return type

InvocationInput

Attributes

InvocationInput.topics

topics DataSelector | list[DataSelector] | None = None #

Topic selectors.

InvocationProvenance

class roboto.domain.actions.invocation_record.InvocationProvenance(/, **data)#

Bases: pydantic.BaseModel

Provenance information for an invocation

Parameters

data Any

Attributes

InvocationProvenance.action

action ActionProvenance #

The Action that was invoked.

InvocationProvenance.executable

executable ExecutableProvenance #

The underlying executable (e.g., Docker image) that was run.

InvocationProvenance.source

source SourceProvenance #

The source of the invocation.

InvocationRecord

class roboto.domain.actions.invocation_record.InvocationRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of an invocation.

Parameters

data Any

Attributes

InvocationRecord.compute_requirements

compute_requirements roboto.domain.actions.action_record.ComputeRequirements #

InvocationRecord.container_parameters

container_parameters roboto.domain.actions.action_record.ContainerParameters #

InvocationRecord.created

created datetime.datetime #

InvocationRecord.data_source

data_source InvocationDataSource #

InvocationRecord.duration

duration datetime.timedelta = None #

InvocationRecord.idempotency_id

idempotency_id str | None = None #

InvocationRecord.input_data

input_data list[str] #

InvocationRecord.invocation_id

invocation_id str #

InvocationRecord.last_heartbeat

last_heartbeat datetime.datetime | None = None #

InvocationRecord.last_status

last_status InvocationStatus #

InvocationRecord.org_id

org_id str #

InvocationRecord.parameter_values

parameter_values dict[str, Any] = None #

InvocationRecord.provenance

provenance InvocationProvenance #

InvocationRecord.rich_input_data

rich_input_data InvocationInput | None = None #

InvocationRecord.status

status list[InvocationStatusRecord] = None #

InvocationRecord.timeout

timeout int #

InvocationRecord.upload_destination

upload_destination InvocationUploadDestination | None = None #

InvocationSource

class roboto.domain.actions.invocation_record.InvocationSource(*args, **kwds)#

Bases: enum.Enum

Method by which an invocation was run

Attributes

InvocationSource.Manual

Manual = 'Manual' #

InvocationSource.Trigger

Trigger = 'Trigger' #

InvocationStatus

class roboto.domain.actions.invocation_record.InvocationStatus#

Bases: int, enum.Enum

Invocation status enum

Attributes

InvocationStatus.Cancelled

Cancelled = 997 #

InvocationStatus.Completed

Completed = 5 #

InvocationStatus.Deadly

Deadly = 999 #

InvocationStatus.Downloading

Downloading = 2 #

InvocationStatus.Failed

Failed = 998 #

InvocationStatus.Processing

Processing = 3 #

InvocationStatus.Queued

Queued = 0 #

InvocationStatus.Scheduled

Scheduled = 1 #

InvocationStatus.Uploading

Uploading = 4 #

InvocationStatus.can_transition_to()

can_transition_to(other)#

Parameters

other InvocationStatus

Return type

bool

InvocationStatus.from_value()

static from_value(v)#

Parameters

v Union[int, str]

Return type

InvocationStatus

InvocationStatus.is_running()

is_running()#

Return type

bool

InvocationStatus.is_terminal()

is_terminal()#

Return type

bool

InvocationStatus.next()

next()#

Return type

Optional[InvocationStatus]

InvocationStatusRecord

class roboto.domain.actions.invocation_record.InvocationStatusRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of an invocation status.

Parameters

data Any

Attributes

InvocationStatusRecord.detail

detail str | None = None #

InvocationStatusRecord.status

status InvocationStatus #

InvocationStatusRecord.timestamp

timestamp datetime.datetime #

InvocationStatusRecord.to_presentable_dict()

to_presentable_dict()#

Return type

dict[str, Optional[str]]

InvocationUploadDestination

class roboto.domain.actions.invocation_record.InvocationUploadDestination(/, **data)#

Bases: pydantic.BaseModel

Default destination to which invocation outputs - if any - should be uploaded.

Specifies where files generated during action execution should be stored. Actions can write files to their output directory, and this destination determines where those files are uploaded after execution completes.

Parameters

data Any

InvocationUploadDestination.dataset()

classmethod dataset(dataset_id)#

Create a dataset upload destination with the given ID.

Parameters

dataset_id str

The ID of the dataset where outputs should be uploaded.

Returns

InvocationUploadDestination

An InvocationUploadDestination configured for the specified dataset.

Attributes

InvocationUploadDestination.destination_id

destination_id str | None = None #

Optional identifier for the upload destination. In the case of a dataset, it would be the dataset ID.

InvocationUploadDestination.destination_type

destination_type UploadDestinationType #

Type of upload destination. By default, outputs are uploaded to a dataset.

Properties

InvocationUploadDestination.is_dataset

is_dataset bool #

True if this is a dataset destination with a dataset ID, False otherwise.

Returns

bool

True if this destination is configured for a dataset with a valid ID.

InvocationUploadDestination.is_unknown

is_unknown bool #

True if the upload destination is not of a supported type, False otherwise.

Return type: bool

InvocationUploadDestination.pre_validate_destination_type()

classmethod pre_validate_destination_type(value)#

Parameters

value Any

Return type

Any

LogRecord

class roboto.domain.actions.invocation_record.LogRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of a log record.

Parameters

data Any

Attributes

LogRecord.log

log str #

LogRecord.partial_id

partial_id str | None = None #

LogRecord.process

process roboto.domain.actions.action_record.ExecutorContainer #

LogRecord.timestamp

timestamp datetime.datetime #

LogsLocation

class roboto.domain.actions.invocation_record.LogsLocation(/, **data)#

Bases: pydantic.BaseModel

Invocation log storage location

Parameters

data Any

Attributes

LogsLocation.bucket

bucket str #

LogsLocation.prefix

prefix str #

SourceProvenance

class roboto.domain.actions.invocation_record.SourceProvenance(/, **data)#

Bases: pydantic.BaseModel

Provenance information for an invocation source

Parameters

data Any

Attributes

SourceProvenance.source_id

source_id str #

SourceProvenance.source_type

source_type InvocationSource #

UploadDestinationType

class roboto.domain.actions.invocation_record.UploadDestinationType(*args, **kwds)#

Bases: enum.Enum

Type of upload destination for invocation outputs.

Defines where files generated by action invocations should be uploaded. Currently supports datasets as the primary destination type.

Attributes

UploadDestinationType.Dataset

Dataset = 'Dataset' #

Outputs will be uploaded to a dataset. This is the default.

UploadDestinationType.Unknown

Unknown = 'Unknown' #

The output destination is unknown.

This destination type exists for compatibility between different versions of the Roboto SDK and the Roboto service backend. It should not be used directly in action invocation requests. If you encounter it in an SDK response, consider upgrading to the latest available SDK version.