roboto.domain.actions.invocation#

Module Contents#

class roboto.domain.actions.invocation.Invocation(record, roboto_client=None)#

An instance of an execution of an action, initiated manually by a user or automatically by a trigger.

An Invocation represents a single execution of an Action with specific inputs, parameters, and configuration. It tracks the execution lifecycle from creation through completion, including status updates, logs, and results.

Invocations are created by calling Action.invoke() or through the UI. They cannot be created directly through the constructor. Each invocation has a unique ID and maintains a complete audit trail of its execution.

Key features:

Status tracking (Queued, Running, Completed, Failed, etc.)
Input data specification and parameter values
Compute requirement and container parameter overrides
Log collection and output file management
Progress monitoring and result retrieval

Parameters:

record (roboto.domain.actions.invocation_record.InvocationRecord)
roboto_client (Optional[roboto.http.RobotoClient])

property action: roboto.domain.actions.invocation_record.ActionProvenance#

Provenance information about the action that was invoked.

Return type:: roboto.domain.actions.invocation_record.ActionProvenance

cancel()#

Cancel this invocation if it is not already in a terminal status.

Attempts to cancel the invocation. If the invocation has already completed, failed, or reached another terminal status, this method has no effect.

Raises:

RobotoNotFoundException – If the invocation is not found.
RobotoUnauthorizedException – If the caller lacks permission to cancel the invocation.

Return type:

None

Examples

Cancel a running invocation:

>>> invocation = Invocation.from_id("iv_12345")
>>> if not invocation.reached_terminal_status:
...     invocation.cancel()

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

The compute requirements (CPU, memory) used for this invocation.

Return type:: roboto.domain.actions.action_record.ComputeRequirements

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

The container parameters used for this invocation.

Return type:: roboto.domain.actions.action_record.ContainerParameters

property created: datetime.datetime#

The timestamp when this invocation was created.

Return type:: datetime.datetime

property current_status: roboto.domain.actions.invocation_record.InvocationStatus#

The current status of this invocation (e.g., Queued, Running, Completed).

Return type:: roboto.domain.actions.invocation_record.InvocationStatus

property data_source: roboto.domain.actions.invocation_record.InvocationDataSource#

The data source that provided input data for this invocation.

Return type:: roboto.domain.actions.invocation_record.InvocationDataSource

property executable: roboto.domain.actions.invocation_record.ExecutableProvenance#

Provenance information about the executable (container) that was run.

Return type:: roboto.domain.actions.invocation_record.ExecutableProvenance

classmethod from_id(invocation_id, roboto_client=None)#

Load an existing invocation by its ID.

Retrieves an invocation from the Roboto platform using its unique identifier.

Parameters:

invocation_id (str) – The unique ID of the invocation to retrieve.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client instance. Uses default if not provided.

Returns:

The Invocation instance.

Raises:

RobotoNotFoundException – If the invocation is not found.
RobotoUnauthorizedException – If the caller lacks permission to access the invocation.

Return type:

Invocation

Examples

Load an invocation and check its status:

>>> invocation = Invocation.from_id("iv_12345")
>>> print(f"Status: {invocation.current_status}")
>>> print(f"Created: {invocation.created}")

get_logs(page_token=None)#

Retrieve runtime STDOUT/STDERR logs generated during this invocation’s execution.

Fetches log records from the invocation’s container execution, with support for pagination to handle large log volumes.

Parameters:

page_token (Optional[str]) – Optional token for pagination. If provided, starts retrieving logs from that point.

Yields:

LogRecord instances containing log messages and metadata.

Raises:

RobotoNotFoundException – If the invocation is not found.
RobotoUnauthorizedException – If the caller lacks permission to access logs.

Return type:

collections.abc.Generator[roboto.domain.actions.invocation_record.LogRecord, None, None]

property id: str#

The unique identifier for this invocation.

Return type:: str

property input_data: roboto.domain.actions.invocation_record.InvocationInput | None#

The input data specification for this invocation, if any.

Return type:: Optional[roboto.domain.actions.invocation_record.InvocationInput]

is_queued_for_scheduling()#

An invocation is queued for scheduling if:: 1. its most recent status is “Queued” 3. and is not “Deadly”

Return type:: bool

property org_id: str#

The organization ID that owns this invocation.

Return type:: str

property parameter_values: dict[str, Any]#

The parameter values that were provided when this invocation was created.

Return type:: dict[str, Any]

classmethod query(spec=None, owner_org_id=None, roboto_client=None)#

Query invocations with optional filtering and pagination.

Searches for invocations based on the provided query specification. Can filter by status, action name, creation time, and other attributes.

Parameters:

spec (Optional[roboto.query.QuerySpecification]) – Query specification with filters, sorting, and pagination. If not provided, returns all accessible invocations.
owner_org_id (Optional[str]) – Organization ID to search within. If not provided, searches in the caller’s organization.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client instance. Uses default if not provided.

Yields:

Invocation instances matching the query criteria.

Raises:

ValueError – If the query specification contains unknown fields.
RobotoUnauthorizedException – If the caller lacks permission to query invocations.

Return type:

collections.abc.Generator[Invocation, None, None]

Examples

Query all invocations:

>>> for invocation in Invocation.query():
...     print(f"Invocation: {invocation.id}")

Query completed invocations:

>>> from roboto.query import QuerySpecification
>>> from roboto.domain.actions import InvocationStatus
>>> spec = QuerySpecification().where("current_status").equals(InvocationStatus.Completed)
>>> completed = list(Invocation.query(spec))

Query recent invocations:

>>> spec = QuerySpecification().order_by("created", ascending=False).limit(10)
>>> recent = list(Invocation.query(spec))

property reached_terminal_status: bool#

True if this invocation has reached a terminal status (Completed, Failed, etc.).

Return type:: bool

property record: roboto.domain.actions.invocation_record.InvocationRecord#

The underlying invocation record containing all invocation data.

Return type:: roboto.domain.actions.invocation_record.InvocationRecord

refresh()#

Return type:: Invocation

set_container_image_digest(digest)#

This is an admin-only operation to memorialize the digest of the container image that was pulled in the course of invoking the action.

Parameters:: digest (str)
Return type:: Invocation

set_logs_location(logs)#

This is an admin-only operation to memorialize the base location where invocation logs are saved.

Use the “get_logs” or “stream_logs” methods to access invocation logs.

Parameters:: logs (roboto.domain.actions.invocation_record.LogsLocation)
Return type:: Invocation

property source: roboto.domain.actions.invocation_record.SourceProvenance#

Provenance information about the source that initiated this invocation.

Return type:: roboto.domain.actions.invocation_record.SourceProvenance

property status_log: list[roboto.domain.actions.invocation_record.InvocationStatusRecord]#

The complete history of status changes for this invocation.

Return type:: list[roboto.domain.actions.invocation_record.InvocationStatusRecord]

stream_logs(last_read=None)#

Parameters:: last_read (Optional[str])
Return type:: collections.abc.Generator[roboto.domain.actions.invocation_record.LogRecord, None, Optional[str]]

property timeout: int#

The timeout in minutes for this invocation.

Return type:: int

to_dict()#

Return type:: dict[str, Any]

update_status(next_status, detail=None)#

Parameters:

next_status (roboto.domain.actions.invocation_record.InvocationStatus)
detail (Optional[str])

Return type:

Invocation

property upload_destination: roboto.domain.actions.invocation_record.InvocationUploadDestination | None#

The destination where output files from this invocation will be uploaded.

Return type:: Optional[roboto.domain.actions.invocation_record.InvocationUploadDestination]

wait_for_terminal_status(timeout=60 * 5, poll_interval=5)#

Wait for the invocation to reach a terminal status.

Throws a TimeoutError if the timeout is reached.

Parameters:

timeout (float) – The maximum amount of time, in seconds, to wait for the invocation to reach a terminal status.
poll_interval (roboto.waiters.Interval) – The amount of time, in seconds, to wait between polling iterations.

Return type:

None