roboto.domain.actions.action#

Module Contents#

class roboto.domain.actions.action.Action(record, roboto_client=None)#

A reusable function to process, transform or analyze data in Roboto.

Actions are containerized functions that can be invoked to process datasets, files, or other data sources within the Roboto platform. They encapsulate processing logic, dependencies, and compute requirements, making data processing workflows reproducible and scalable.

Actions can be created, updated, invoked, and managed through this class. They support parameterization, inheritance from other actions, and can be triggered automatically based on events or conditions.

An Action consists of:

  • Container image and execution parameters

  • Input/output specifications

  • Compute requirements (CPU, memory, etc.)

  • Parameters that can be customized at invocation time

  • Metadata and tags for organization

Actions are owned by organizations and can have different accessibility levels (private to organization or public in the Action Hub).

Parameters:
property accessibility: roboto.domain.actions.action_record.Accessibility#

The accessibility level of this action (Organization or ActionHub).

Return type:

roboto.domain.actions.action_record.Accessibility

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

The compute requirements (CPU, memory) for running this action.

Return type:

Optional[roboto.domain.actions.action_record.ComputeRequirements]

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

The container parameters including image URI and execution settings.

Return type:

Optional[roboto.domain.actions.action_record.ContainerParameters]

classmethod create(name, compute_requirements=None, container_parameters=None, description=None, inherits=None, metadata=None, parameters=None, requires_downloaded_inputs=None, short_description=None, tags=None, timeout=None, uri=None, caller_org_id=None, roboto_client=None)#

Create a new action in the Roboto platform.

Creates a new action with the specified configuration. The action will be owned by the caller’s organization and can be invoked to process data.

Parameters:

  • name (str) – Unique name for the action within the organization.

  • compute_requirements (Optional[roboto.domain.actions.action_record.ComputeRequirements]) – CPU, memory, and other compute specifications.

  • container_parameters (Optional[roboto.domain.actions.action_record.ContainerParameters]) – Container image URI, entrypoint, and environment variables.

  • description (Optional[str]) – Detailed description of what the action does.

  • inherits (Optional[roboto.domain.actions.action_record.ActionReference]) – Reference to another action to inherit configuration from.

  • metadata (Optional[dict[str, Any]]) – Custom key-value metadata to associate with the action.

  • parameters (Optional[list[roboto.domain.actions.action_record.ActionParameter]]) – List of parameters that can be provided at invocation time.

  • requires_downloaded_inputs (Optional[bool]) – Whether input files should be downloaded before execution.

  • short_description (Optional[str]) – Brief description (max 140 characters) for display purposes.

  • tags (Optional[list[str]]) – List of tags for categorizing and searching actions.

  • timeout (Optional[int]) – Maximum execution time in minutes before the action is terminated.

  • uri (Optional[str]) – Container image URI if not inheriting from another action.

  • caller_org_id (Optional[str]) – Organization ID to create the action in. Defaults to caller’s org.

  • roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client instance. Uses default if not provided.

Returns:

The newly created Action instance.

Raises:
Return type:

Action

Examples

Create a simple action:

>>> action = Action.create(
...     name="hello_world",
...     uri="ubuntu:latest",
...     description="A simple hello world action"
... )

Create an action with parameters and compute requirements:

>>> from roboto.domain.actions import ComputeRequirements, ActionParameter
>>> action = Action.create(
...     name="data_processor",
...     uri="my-registry.com/processor:v1.0",
...     description="Processes sensor data with configurable parameters",
...     compute_requirements=ComputeRequirements(vCPU=4096, memory=8192),
...     parameters=[
...         ActionParameter(name="threshold", required=True, description="Processing threshold"),
...         ActionParameter(name="output_format", default="json", description="Output format")
...     ],
...     tags=["data-processing", "sensors"],
...     timeout=60
... )

Create an action that inherits from another:

>>> base_action = Action.from_name("base_processor", owner_org_id="roboto-public")
>>> derived_action = Action.create(
...     name="custom_processor",
...     inherits=base_action.record.reference,
...     description="Custom processor based on base_processor",
...     metadata={"version": "2.0", "team": "data-science"}
... )
property created: datetime.datetime#

The timestamp when this action was created.

Return type:

datetime.datetime

property created_by: str#

The user ID who created this action.

Return type:

str

delete()#

Delete this action from the Roboto platform.

Permanently removes this action and all its versions. This operation cannot be undone.

Raises:
Return type:

None

Examples

Delete an action:

>>> action = Action.from_name("old_action")
>>> action.delete()
property description: str | None#

The detailed description of what this action does.

Return type:

Optional[str]

property digest: str#

The unique digest identifying this specific version of the action.

Return type:

str

classmethod from_name(name, digest=None, owner_org_id=None, roboto_client=None)#

Load an existing action by name.

Retrieves an action from the Roboto platform by its name and optionally a specific version digest. Action names are unique within an organization, so a name + org_id combination always provides a fully qualified reference to a specific action.

Parameters:

  • name (str) – Name of the action to retrieve. Must be unique within the organization.

  • digest (Optional[str]) – Specific version digest of the action. If not provided, returns the latest version.

  • owner_org_id (Optional[str]) – Organization ID that owns the action. If not provided, searches in the caller’s organization.

  • roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client instance. Uses default if not provided.

Returns:

The Action instance.

Raises:
Return type:

Action

Examples

Load the latest version of an action:

>>> action = Action.from_name("data_processor")

Load a specific version of an action:

>>> action = Action.from_name(
...     "data_processor",
...     digest="abc123def456"
... )

Load an action from another organization:

>>> action = Action.from_name(
...     "public_processor",
...     owner_org_id="roboto-public"
... )
property inherits_from: roboto.domain.actions.action_record.ActionReference | None#

Reference to another action this action inherits configuration from.

Return type:

Optional[roboto.domain.actions.action_record.ActionReference]

invoke(invocation_source, data_source_id=None, data_source_type=None, input_data=None, upload_destination=None, compute_requirement_overrides=None, container_parameter_overrides=None, idempotency_id=None, invocation_source_id=None, parameter_values=None, timeout=None, caller_org_id=None)#

Invokes this action using any inputs and options provided.

Executes this action with the specified parameters and returns an Invocation object that can be used to track progress and retrieve results.

Parameters:
Returns:

An Invocation object that can be used to track the invocation’s progress.

Raises:
Return type:

roboto.domain.actions.invocation.Invocation

Examples

Basic invocation with a dataset:

>>> from roboto import Action, InvocationSource
>>> action = Action.from_name("ros_ingestion", owner_org_id="roboto-public")
>>> iv = action.invoke(
...     invocation_source=InvocationSource.Manual,
...     data_source_id="ds_12345",
...     data_source_type=InvocationDataSourceType.Dataset,
...     input_data=["**/*.bag"],
...     upload_destination=InvocationUploadDestination.dataset("ds_12345")
... )
>>> iv.wait_for_terminal_status()

Invocation with compute requirement overrides:

>>> from roboto import Action, InvocationSource, ComputeRequirements
>>> action = Action.from_name("image_processing", owner_org_id="roboto-public")
>>> compute_reqs = ComputeRequirements(vCPU=4096, memory=8192)
>>> iv = action.invoke(
...     invocation_source=InvocationSource.Manual,
...     compute_requirement_overrides=compute_reqs,
...     parameter_values={"threshold": 0.75}
... )
>>> status = iv.wait_for_terminal_status()
>>> print(status)
'COMPLETED'
property metadata: dict[str, Any]#

Custom metadata key-value pairs associated with this action.

Return type:

dict[str, Any]

property modified: datetime.datetime#

The timestamp when this action was last modified.

Return type:

datetime.datetime

property modified_by: str#

The user ID who last modified this action.

Return type:

str

property name: str#

The unique name of this action within its organization.

Return type:

str

property org_id: str#

The organization ID that owns this action.

Return type:

str

property parameters: collections.abc.Sequence[roboto.domain.actions.action_record.ActionParameter]#

The list of parameters that can be provided when invoking this action.

Return type:

collections.abc.Sequence[roboto.domain.actions.action_record.ActionParameter]

property published: datetime.datetime | None#

The timestamp when this action was published to the Action Hub, if applicable.

Return type:

Optional[datetime.datetime]

classmethod query(spec=None, accessibility=Accessibility.Organization, owner_org_id=None, roboto_client=None)#

Query actions with optional filtering and pagination.

Searches for actions based on the provided query specification. Can search within an organization or across the public Action Hub.

Parameters:

  • spec (Optional[roboto.query.QuerySpecification]) – Query specification with filters, sorting, and pagination. If not provided, returns all accessible actions.

  • accessibility (roboto.domain.actions.action_record.Accessibility) – Whether to search organization actions or public Action Hub. Defaults to Organization.

  • 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:

Action instances matching the query criteria.

Raises:

  • ValueError – If the query specification contains unknown fields.

  • RobotoUnauthorizedException – If the caller lacks permission to query actions.

Return type:

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

Examples

Query all actions in your organization:

>>> for action in Action.query():
...     print(f"Action: {action.name}")

Query actions with specific tags:

>>> from roboto.query import QuerySpecification
>>> spec = QuerySpecification().where("tags").contains("ml")
>>> for action in Action.query(spec):
...     print(f"ML Action: {action.name}")

Query public actions in the Action Hub:

>>> from roboto.domain.actions import Accessibility
>>> spec = QuerySpecification().where("name").contains("ros")
>>> for action in Action.query(spec, accessibility=Accessibility.ActionHub):
...     print(f"Public ROS Action: {action.name}")

Query with pagination:

>>> spec = QuerySpecification().limit(10).order_by("created", ascending=False)
>>> recent_actions = list(Action.query(spec))
>>> print(f"Found {len(recent_actions)} recent actions")
property record: roboto.domain.actions.action_record.ActionRecord#

The underlying action record containing all action data.

Return type:

roboto.domain.actions.action_record.ActionRecord

property requires_downloaded_inputs: bool#

Whether input files should be downloaded before executing this action.

Return type:

bool

set_accessibility(accessibility)#

Set the accessibility level of this action.

Changes whether this action is private to the organization or published to the public Action Hub.

Parameters:

accessibility (roboto.domain.actions.action_record.Accessibility) – The new accessibility level (Organization or ActionHub).

Returns:

This Action instance with updated accessibility.

Raises:

RobotoUnauthorizedException – If the caller lacks permission to modify the action.

Return type:

Action

Examples

Make an action public in the Action Hub:

>>> from roboto.domain.actions import Accessibility
>>> action = Action.from_name("my_action")
>>> action.set_accessibility(Accessibility.ActionHub)

Make an action private to the organization:

>>> action.set_accessibility(Accessibility.Organization)
property short_description: str | None#

A brief description of the action (max 140 characters) for display purposes.

Return type:

Optional[str]

property tags: list[str]#

The list of tags associated with this action for categorization.

Return type:

list[str]

property timeout: int | None#

The maximum execution time in minutes before the action is terminated.

Return type:

Optional[int]

to_dict()#

Convert this action to a dictionary representation.

Returns:

Dictionary containing all action data in JSON-serializable format.

Return type:

dict[str, Any]

Examples

Get action as dictionary:

>>> action = Action.from_name("my_action")
>>> action_dict = action.to_dict()
>>> print(action_dict["name"])
'my_action'
update(compute_requirements=NotSet, container_parameters=NotSet, description=NotSet, inherits=NotSet, metadata_changeset=NotSet, parameter_changeset=NotSet, short_description=NotSet, timeout=NotSet, uri=NotSet, requires_downloaded_inputs=NotSet)#

Update this action with new configuration.

Updates the action with the provided changes. Only specified parameters will be modified; others remain unchanged. This creates a new version of the action.

Parameters:
Returns:

This Action instance with updated configuration.

Raises:
Return type:

Action

Examples

Update action description and timeout:

>>> action = Action.from_name("my_action")
>>> action.update(
...     description="Updated description of what this action does",
...     timeout=45
... )

Update compute requirements:

>>> from roboto.domain.actions import ComputeRequirements
>>> action.update(
...     compute_requirements=ComputeRequirements(vCPU=4096, memory=8192)
... )

Add metadata using changeset:

>>> from roboto.updates import MetadataChangeset
>>> changeset = MetadataChangeset().set("version", "2.0").set("team", "ml")
>>> action.update(metadata_changeset=changeset)
property uri: str | None#

The container image URI for this action.

Return type:

Optional[str]

roboto.domain.actions.action.SHORT_DESCRIPTION_LENGTH = 140#
On this page