roboto.domain.actions.invocation_record#

Module Contents#

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

Bases: pydantic.BaseModel

Provenance information for an action

Parameters:: data (Any)

digest: str | None = None#

name: str#

org_id: str#

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

Bases: pydantic.BaseModel

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

Parameters:: data (Any)

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

ensure_not_empty()#

Return type:: DataSelector

ids: list[str] | None = None#: Specific input IDs (e.g. a dataset ID).

names: list[str] | None = None#: Specific input names (e.g. topic names).

query: str | None = None#: RoboQL query representing the desired inputs.

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

Bases: pydantic.BaseModel

Provenance information for an action executable

Parameters:: data (Any)

container_image_digest: str | None = None#

container_image_uri: str | None = None#

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)

ensure_not_empty()#

Return type:: FileSelector

paths: list[str] | None = None#

path LIKE <path pattern in double quotes>

Type:: File paths or patterns. Prefer RoboQL

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)

data_source_id: str#: The ID of the data source. For Dataset type, this is a dataset ID.

data_source_type: InvocationDataSourceType#: The type of data source (currently only Dataset).

is_unspecified()#

Check if this data source is unspecified.

Returns:: True if this is an unspecified data source, False otherwise.
Return type:: bool

static unspecified()#

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

Returns:: An InvocationDataSource instance representing an unspecified data source.
Return type:: InvocationDataSource

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.

Dataset = 'Dataset'#

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)

ensure_not_empty()#

Return type:: InvocationInput

property file_paths: list[str]#

Return type:: list[str]

classmethod file_query(roboql_query)#

Specify file inputs using a RoboQL query.

Parameters:: roboql_query (str)
Return type:: InvocationInput

files: FileSelector | list[FileSelector] | None = None#: File selectors.

classmethod from_dataset_file_paths(dataset_id, file_paths)#

Parameters:

dataset_id (str)
file_paths (list[str])

Return type:

InvocationInput

property safe_files: list[FileSelector]#

Return type:: list[FileSelector]

property safe_topics: list[DataSelector]#

Return type:: list[DataSelector]

classmethod topic_query(roboql_query)#

Specify topic inputs using a RoboQL query.

Parameters:: roboql_query (str)
Return type:: InvocationInput

topics: DataSelector | list[DataSelector] | None = None#: Topic selectors.

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

Bases: pydantic.BaseModel

Provenance information for an invocation

Parameters:: data (Any)

action: ActionProvenance#: The Action that was invoked.

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

source: SourceProvenance#: The source of the invocation.

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

Bases: pydantic.BaseModel

A wire-transmissible representation of an invocation.

Parameters:: data (Any)

compute_requirements: roboto.domain.actions.action_record.ComputeRequirements#

container_parameters: roboto.domain.actions.action_record.ContainerParameters#

created: datetime.datetime#

data_source: InvocationDataSource#

duration: datetime.timedelta = None#

idempotency_id: str | None = None#

input_data: list[str]#

invocation_id: str#

last_heartbeat: datetime.datetime | None = None#

last_status: InvocationStatus#

org_id: str#

parameter_values: dict[str, Any] = None#

provenance: InvocationProvenance#

rich_input_data: InvocationInput | None = None#

status: list[InvocationStatusRecord] = None#

timeout: int#

upload_destination: InvocationUploadDestination | None = None#

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

Bases: enum.Enum

Method by which an invocation was run

Manual = 'Manual'#

Trigger = 'Trigger'#

class roboto.domain.actions.invocation_record.InvocationStatus#

Bases: int, enum.Enum

Invocation status enum

Cancelled = 997#

Completed = 5#

Deadly = 999#

Downloading = 2#

Failed = 998#

Processing = 3#

Queued = 0#

Scheduled = 1#

Uploading = 4#

can_transition_to(other)#

Parameters:: other (InvocationStatus)
Return type:: bool

static from_value(v)#

Parameters:: v (Union[int, str])
Return type:: InvocationStatus

is_running()#

Return type:: bool

is_terminal()#

Return type:: bool

next()#

Return type:: Optional[InvocationStatus]

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

Bases: pydantic.BaseModel

A wire-transmissible representation of an invocation status.

Parameters:: data (Any)

detail: str | None = None#

status: InvocationStatus#

timestamp: datetime.datetime#

to_presentable_dict()#

Return type:: dict[str, Optional[str]]

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)

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:: An InvocationUploadDestination configured for the specified dataset.
Return type:: InvocationUploadDestination

destination_id: str | None = None#: Optional identifier for the upload destination. In the case of a dataset, it would be the dataset ID.

destination_type: UploadDestinationType#: Type of upload destination. By default, outputs are uploaded to a dataset.

property is_dataset: bool#

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

Returns:: True if this destination is configured for a dataset with a valid ID.
Return type:: bool

property is_unknown: bool#

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

Return type:: bool

classmethod pre_validate_destination_type(value)#

Parameters:: value (Any)
Return type:: Any

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

Bases: pydantic.BaseModel

A wire-transmissible representation of a log record.

Parameters:: data (Any)

log: str#

partial_id: str | None = None#

process: roboto.domain.actions.action_record.ExecutorContainer#

timestamp: datetime.datetime#

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

Bases: pydantic.BaseModel

Invocation log storage location

Parameters:: data (Any)

bucket: str#

prefix: str#

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

Bases: pydantic.BaseModel

Provenance information for an invocation source

Parameters:: data (Any)

source_id: str#

source_type: InvocationSource#

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.

Dataset = 'Dataset'#: Outputs will be uploaded to a dataset. This is the default.

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.