--- sidebar: hidden: true title: roboto.domain.actions.action --- ## Module Contents ### Action ```python class roboto.domain.actions.action.Action( record: roboto.domain.actions.action_record.ActionRecord, roboto_client: Optional[roboto.http.RobotoClient] = None, ) ``` `from roboto import Action` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L53-L745) 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** - **record** (`roboto.domain.actions.action_record.ActionRecord`) - **roboto_client** (`Optional[roboto.http.RobotoClient]`) **Properties** - **Action.accessibility** (`roboto.domain.actions.action_record.Accessibility`): The accessibility level of this action (Organization or ActionHub). - **Action.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]` - **Action.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]` - **Action.created** (`datetime.datetime`): The timestamp when this action was created. - **Action.created_by** (`str`): The user ID who created this action. - **Action.description** (`str | None`): The detailed description of what this action does. Return type: `Optional[str]` - **Action.digest** (`str`): The unique digest identifying this specific version of the action. - **Action.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]` - **Action.metadata** (`dict[str, Any]`): Custom metadata key-value pairs associated with this action. - **Action.modified** (`datetime.datetime`): The timestamp when this action was last modified. - **Action.modified_by** (`str`): The user ID who last modified this action. - **Action.name** (`str`): The unique name of this action within its organization. - **Action.org_id** (`str`): The organization ID that owns this action. - **Action.parameters** (`collections.abc.Sequence[roboto.domain.actions.action_record.ActionParameter]`): The list of parameters that can be provided when invoking this action. - **Action.published** (`datetime.datetime | None`): The timestamp when this action was published to the Action Hub, if applicable. Return type: `Optional[datetime.datetime]` - **Action.record** (`roboto.domain.actions.action_record.ActionRecord`): The underlying action record containing all action data. - **Action.requires_downloaded_inputs** (`bool`): Whether input files should be downloaded before executing this action. - **Action.short_description** (`str | None`): A brief description of the action (max 140 characters) for display purposes. Return type: `Optional[str]` - **Action.tags** (`list[str]`): The list of tags associated with this action for categorization. - **Action.timeout** (`int | None`): The maximum execution time in minutes before the action is terminated. Return type: `Optional[int]` - **Action.uri** (`str | None`): The container image URI for this action. Return type: `Optional[str]` #### Action.create() ```python @classmethod def create( name: str, compute_requirements: Optional[roboto.domain.actions.action_record.ComputeRequirements] = None, container_parameters: Optional[roboto.domain.actions.action_record.ContainerParameters] = None, description: Optional[str] = None, inherits: Optional[roboto.domain.actions.action_record.ActionReference] = None, metadata: Optional[dict[str, Any]] = None, parameters: Optional[list[roboto.domain.actions.action_record.ActionParameter]] = None, requires_downloaded_inputs: Optional[bool] = None, short_description: Optional[str] = None, tags: Optional[list[str]] = None, timeout: Optional[int] = None, uri: Optional[str] = None, caller_org_id: Optional[str] = None, roboto_client: Optional[roboto.http.RobotoClient] = None, ) -> Action ``` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L81-L193) 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** - `Action`: The newly created Action instance. **Raises** - [`RobotoIllegalArgumentException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoIllegalArgumentException): If short_description exceeds 140 characters or other validation errors occur. - [`RobotoInvalidRequestException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoInvalidRequestException): If the request is malformed. - [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): If the caller lacks permission to create actions. **Usage** Create a simple action: ```python action = Action.create( name="hello_world", uri="ubuntu:latest", description="A simple hello world action" ) ``` Create an action with parameters and compute requirements: ```python 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: ```python 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"}, ) ``` #### Action.delete() ```python def delete() -> None ``` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L461-L480) Delete this action from the Roboto platform. Permanently removes this action and all its versions. This operation cannot be undone. **Raises** - [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): If the action is not found. - [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): If the caller lacks permission to delete the action. **Returns** - `None` **Usage** Delete an action: ```python action = Action.from_name("old_action") action.delete() ``` #### Action.from_name() ```python @classmethod def from_name( name: str, digest: Optional[str] = None, owner_org_id: Optional[str] = None, roboto_client: Optional[roboto.http.RobotoClient] = None, ) -> Action ``` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L196-L246) 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** - `Action`: The Action instance. **Raises** - [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): If the action is not found. - [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): If the caller lacks permission to access the action. **Usage** Load the latest version of an action: ```python action = Action.from_name("data_processor") ``` Load a specific version of an action: ```python action = Action.from_name("data_processor", digest="abc123def456") ``` Load an action from another organization: ```python action = Action.from_name("public_processor", owner_org_id="roboto-public") ``` #### Action.invoke() ```python def invoke( invocation_source: roboto.domain.actions.invocation_record.InvocationSource, data_source_id: Optional[str] = None, data_source_type: Optional[roboto.domain.actions.invocation_record.InvocationDataSourceType] = None, input_data: Optional[Union[list[str], roboto.domain.actions.invocation_record.InvocationInput]] = None, upload_destination: Optional[roboto.domain.actions.invocation_record.InvocationUploadDestination] = None, compute_requirement_overrides: Optional[roboto.domain.actions.action_record.ComputeRequirements] = None, container_parameter_overrides: Optional[roboto.domain.actions.action_record.ContainerParameters] = None, idempotency_id: Optional[str] = None, invocation_source_id: Optional[str] = None, parameter_values: Optional[dict[str, Any]] = None, timeout: Optional[int] = None, caller_org_id: Optional[str] = None, ) -> roboto.domain.actions.invocation.Invocation ``` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L482-L604) 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** - **invocation_source** (`roboto.domain.actions.invocation_record.InvocationSource`): Manual, trigger, etc. - **data_source_type** (`Optional[roboto.domain.actions.invocation_record.InvocationDataSourceType]`): If set, should equal Dataset for backward compatibility. - **data_source_id** (`Optional[str]`): If set, should be a dataset ID for backward compatibility. - **input_data** (`Optional[Union[list[str], roboto.domain.actions.invocation_record.InvocationInput]]`): Either a list of file name patterns, or an InvocationInput specification. - **upload_destination** (`Optional[roboto.domain.actions.invocation_record.InvocationUploadDestination]`): Default upload destination (e.g. dataset) for files written to the invocation's output directory. - **compute_requirement_overrides** (`Optional[roboto.domain.actions.action_record.ComputeRequirements]`): Overrides for the action's default compute requirements (e.g. vCPU) - **container_parameter_overrides** (`Optional[roboto.domain.actions.action_record.ContainerParameters]`): Overrides for the action's default container parameters (e.g. entrypoint) - **idempotency_id** (`Optional[str]`): Unique ID to ensure an invocation is run exactly once. - **invocation_source_id** (`Optional[str]`): ID of the trigger or manual operator performing the invocation. - **parameter_values** (`Optional[dict[str, Any]]`): Action parameter values. - **timeout** (`Optional[int]`): Action timeout in minutes. - **caller_org_id** (`Optional[str]`): Org ID of the caller. **Returns** - `roboto.domain.actions.invocation.Invocation`: An Invocation object that can be used to track the invocation's progress. **Raises** - [`RobotoIllegalArgumentException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoIllegalArgumentException): Invalid method parameters or combinations. - [`RobotoInvalidRequestException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoInvalidRequestException): Incorrectly formed request. - [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): The caller is not authorized to invoke this action. **Usage** Basic invocation with a dataset: ```python 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: ```python 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' ``` #### Action.query() ```python @classmethod def query( spec: Optional[roboto.query.QuerySpecification] = None, accessibility: roboto.domain.actions.action_record.Accessibility = Accessibility.Organization, owner_org_id: Optional[str] = None, roboto_client: Optional[roboto.http.RobotoClient] = None, ) -> collections.abc.Generator[Action, None, None] ``` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L249-L336) 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`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): If the caller lacks permission to query actions. **Returns** - `collections.abc.Generator[Action, None, None]` **Usage** Query all actions in your organization: ```python for action in Action.query(): print(f"Action: {action.name}") ``` Query actions with specific tags: ```python 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: ```python 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: ```python spec = QuerySpecification().limit(10).order_by("created", ascending=False) recent_actions = list(Action.query(spec)) print(f"Found {len(recent_actions)} recent actions") ``` #### Action.set_accessibility() ```python def set_accessibility( accessibility: roboto.domain.actions.action_record.Accessibility, ) -> Action ``` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L606-L640) 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** - `Action`: This Action instance with updated accessibility. **Raises** - [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): If the caller lacks permission to modify the action. **Usage** Make an action public in the Action Hub: ```python from roboto.domain.actions import Accessibility action = Action.from_name("my_action") action.set_accessibility(Accessibility.ActionHub) ``` Make an action private to the organization: ```python action.set_accessibility(Accessibility.Organization) ``` #### Action.to_dict() ```python def to_dict() -> dict[str, Any] ``` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L642-L656) Convert this action to a dictionary representation. **Returns** - `dict[str, Any]`: Dictionary containing all action data in JSON-serializable format. **Usage** Get action as dictionary: ```python action = Action.from_name("my_action") action_dict = action.to_dict() print(action_dict["name"]) # 'my_action' ``` #### Action.update() ```python def update( compute_requirements: Optional[Union[roboto.domain.actions.action_record.ComputeRequirements, roboto.sentinels.NotSetType]] = NotSet, container_parameters: Optional[Union[roboto.domain.actions.action_record.ContainerParameters, roboto.sentinels.NotSetType]] = NotSet, description: Optional[Union[str, roboto.sentinels.NotSetType]] = NotSet, inherits: Optional[Union[roboto.domain.actions.action_record.ActionReference, roboto.sentinels.NotSetType]] = NotSet, metadata_changeset: Union[roboto.updates.MetadataChangeset, roboto.sentinels.NotSetType] = NotSet, parameter_changeset: Union[roboto.domain.actions.action_record.ActionParameterChangeset, roboto.sentinels.NotSetType] = NotSet, short_description: Optional[Union[str, roboto.sentinels.NotSetType]] = NotSet, timeout: Optional[Union[int, roboto.sentinels.NotSetType]] = NotSet, uri: Optional[Union[str, roboto.sentinels.NotSetType]] = NotSet, requires_downloaded_inputs: Union[bool, roboto.sentinels.NotSetType] = NotSet, ) -> Action ``` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L658-L745) 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** - **compute_requirements** (`Optional[Union[roboto.domain.actions.action_record.ComputeRequirements, roboto.sentinels.NotSetType]]`): New compute requirements (CPU, memory). - **container_parameters** (`Optional[Union[roboto.domain.actions.action_record.ContainerParameters, roboto.sentinels.NotSetType]]`): New container parameters (image, entrypoint, etc.). - **description** (`Optional[Union[str, roboto.sentinels.NotSetType]]`): New detailed description. - **inherits** (`Optional[Union[roboto.domain.actions.action_record.ActionReference, roboto.sentinels.NotSetType]]`): New action reference to inherit from. - **metadata_changeset** (`Union[roboto.updates.MetadataChangeset, roboto.sentinels.NotSetType]`): Changes to apply to metadata (add, remove, update keys). - **parameter_changeset** (`Union[roboto.domain.actions.action_record.ActionParameterChangeset, roboto.sentinels.NotSetType]`): Changes to apply to parameters (add, remove, update). - **short_description** (`Optional[Union[str, roboto.sentinels.NotSetType]]`): New brief description (max 140 characters). - **timeout** (`Optional[Union[int, roboto.sentinels.NotSetType]]`): New maximum execution time in minutes. - **uri** (`Optional[Union[str, roboto.sentinels.NotSetType]]`): New container image URI. - **requires_downloaded_inputs** (`Union[bool, roboto.sentinels.NotSetType]`): Whether to download input files before execution. **Returns** - `Action`: This Action instance with updated configuration. **Raises** - [`RobotoIllegalArgumentException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoIllegalArgumentException): If short_description exceeds 140 characters or other validation errors occur. - [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): If the caller lacks permission to update the action. **Usage** Update action description and timeout: ```python action = Action.from_name("my_action") action.update(description="Updated description of what this action does", timeout=45) ``` Update compute requirements: ```python from roboto.domain.actions import ComputeRequirements action.update(compute_requirements=ComputeRequirements(vCPU=4096, memory=8192)) ``` Add metadata using changeset: ```python from roboto.updates import MetadataChangeset changeset = MetadataChangeset().set("version", "2.0").set("team", "ml") action.update(metadata_changeset=changeset) ``` ### SHORT_DESCRIPTION_LENGTH ```python roboto.domain.actions.action.SHORT_DESCRIPTION_LENGTH = 140 ``` `from roboto.domain.actions.action import SHORT_DESCRIPTION_LENGTH` [Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/actions/action.py#L50-L50)