roboto.action_runtime#

Submodules#

Package Contents#

class roboto.action_runtime.ActionInput#

Resolved references to input data an Action was given to operate on.

To use, access via get_input().

Example

From within an Action, list files passed as input and check their size:

>>> action_runtime = ActionRuntime.from_env()
>>> action_input = action_runtime.get_input()
>>> for file, local_path in action_input.files:
>>>     print(f"{file.file_id} is {local_path.stat().st_size} bytes")
files: collections.abc.Sequence[tuple[roboto.domain.files.File, pathlib.Path | None]] = []#

Files passed as input data to an action invocation.

A file is represented as a tuple of (File, Optional[Path]) where: - File exposes metadata about the file and useful file operations - Optional[Path] is the local file path if the file has been downloaded

classmethod from_record(record, roboto_client)#

Create an ActionInput instance from its serialized representation.

Parameters:
Return type:

ActionInput

get_topics_by_name(topic_name)#

Return any topics in this ActionInput that have the provided name.

Parameters:

topic_name (str) – Topic name to look for.

Returns:

A list of matching Topic instances from self.topics. If no topics have the provided name, the list will be empty. Otherwise, there will be one or more topics in the list, depending on the topic selectors provided to the action invocation.

Return type:

list[roboto.domain.topics.Topic]

topics: collections.abc.Sequence[roboto.domain.topics.Topic] = []#

Topics passed as input data to an action invocation.

class roboto.action_runtime.ActionRuntime(dataset_id, input_dir, invocation_id, org_id, output_dir, roboto_env, roboto_client=None)#

A utility for performing common lookups and other operations during a Roboto action’s runtime. The easiest and most common way to initialize this is:

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()

…which will inspect environment variables to initialize the ActionRuntime.

If you want to test a script using ActionRuntime in a setting such as a developer machine or unit test, and you don’t want to set environment variables to mirror Roboto’s remote execution environment, you can also initialize ActionRuntime directly like:

>>> import pathlib
>>> from roboto import ActionRuntime
>>>
>>> action_runtime = ActionRuntime(
...     dataset_id="ds_XXXXXXXXXXXX",
...     input_dir=pathlib.Path("/path/to/local/input/dir"),
...     invocation_id="iv_XXXXXXXXXXXX",
...     org_id="og_XXXXXXXXXXXX",
...     output_dir=pathlib.Path("/path/to/local/output/dir"),
... )
Parameters:
property dataset: roboto.domain.datasets.Dataset#

A Dataset object for the dataset whose data this action is operating on.

This object will be lazy-initialized the first time it is accessed, which might result in a RobotoNotFoundException if the dataset does not exist. After the first call, the dataset will be cached.

This is particularly useful for adding tags or metadata to a dataset at runtime, for example:

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> action_runtime.dataset.put_tags(["tagged_by_action"])
>>> action_runtime.dataset.put_metadata({"voltage_spikes_seen": 693})
Return type:

roboto.domain.datasets.Dataset

property dataset_id: str#

The ID of the dataset whose data this action is operating on.

Return type:

str

property file_changeset_manager: roboto.action_runtime.file_changeset.FilesChangesetFileManager#

A FilesChangesetFileManager which can be used to associate tags and metadata with the yet-to-be-uploaded files in this invocation’s output directory. In practice, you might use this like:

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>>
>>> my_output_file = action_runtime.output_dir / "nested" / "my_output_file.txt"
>>> my_output_file.write_text("Hello World")
>>>
>>> action_runtime.file_changeset_manager.put_tags("nested/my_output_file.txt", ["tagged_by_action"])
>>> action_runtime.file_changeset_manager.put_fields(
... "nested/my_output_file.txt", {"roboto_proficiency": "extreme - I can annotate output files!"}
... )

This only works for files that don’t exist yet. To tag existing files (such as files in the input directory), you should instead use:

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>>
>>> existing_file = action_runtime.dataset.get_file_by_path("some_file_that_already_exists.txt")
>>> existing_file.put_tags(["tagged_by_action"])
>>> existing_file.put_metadata({"roboto_proficiency": "also extreme - I can annotate input files!"})

For more info, see the top-level docs on the FilesChangesetFileManager class.

Return type:

roboto.action_runtime.file_changeset.FilesChangesetFileManager

classmethod from_env()#

Initializes an ActionRuntime based on values in environment variables. This will throw an exception if any required environment variables are not available. All of these will be available at runtime in Roboto’s remote execution environment.

Example

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
Return type:

ActionRuntime

get_input()#

Instance of ActionInput containing resolved references to input data.

Return type:

roboto.action_runtime.action_input.ActionInput

get_parameter(name)#

Gets the value of the action parameter with the given name.

Parameters:

name (str)

Return type:

str

property input_dir: pathlib.Path#

The directory where the action’s input files are located.

Return type:

pathlib.Path

property invocation: roboto.domain.actions.Invocation#

An Invocation object for the currently running action invocation.

This object will be lazy-initialized the first time it is accessed, which might result in a RobotoNotFoundException if the invocation does not exist. After the first call, the invocation will be cached.

Return type:

roboto.domain.actions.Invocation

property invocation_id: str#

The ID of the currently running action invocation.

Return type:

str

property org: roboto.domain.orgs.Org#

An Org object for the org which invoked the currently running action.

This object will be lazy-initialized the first time it is accessed, which might result in a RobotoNotFoundException if the org does not exist. After the first call, the org will be cached.

Return type:

roboto.domain.orgs.Org

property org_id: str#

The ID of the org which invoked the currently running action.

Return type:

str

property output_dir: pathlib.Path#

The directory where the action’s output files are expected. After the user portion of the action runtime concludes (i.e. when their container exits with a 0 exit code), every file in this directory will be uploaded to the dataset associated with this action invocation.

Return type:

pathlib.Path

property roboto_client: roboto.http.RobotoClient#

The RobotoClient instance used by this action runtime.

Return type:

roboto.http.RobotoClient

class roboto.action_runtime.FilesChangesetFileManager#

This class is used to pre-write tags/metadata updates to files which haven’t been uploaded yet, but will be at the conclusion of an action, by virtue of being in that action’s output directory.

It uses a “file changeset” file to accumulate these pending updates during an action’s runtime, and then applies them automatically at the end of an action, after the action’s output directory has been uploaded.

The most common way to get access to this would be via roboto.action_runtime.ActionRuntime.

put_fields(relative_path, metadata)#

Adds metadata key/value pairs to a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime.

This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.put_fields("images/front0_raw_000734.jpg", {"cars": 2, "trucks": 3})
>>>
>>> # Actually there was a 3rd car I missed in the first pass, and a plane, let me fix that...
>>> file_changeset_manager.put_fields("images/front0_raw_000734.jpg", {"cars": 3, "planes": 1})
Parameters:

  • relative_path (str)

  • metadata (dict[str, Any])

put_tags(relative_path, tags)#

Adds tags to a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime.

This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.put_tags("images/front0_raw_000734.jpg", ["cloudy", "rainy"]})
Parameters:

  • relative_path (str)

  • tags (list[str])

remove_fields(relative_path, keys)#

Removes metadata key/value pairs from a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime. You’ll generally only need this to remove values which were added by a previous call to put_fields().

This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.put_fields("images/front0_raw_000734.jpg", {"cars": 2, "trucks": 3})
>>>
>>> # Whoops, actually I don't want to count those trucks...
>>> file_changeset_manager.remove_fields("images/front0_raw_000734.jpg", ["trucks"])
Parameters:

  • relative_path (str)

  • keys (list[str])

remove_tags(relative_path, tags)#

Removes tags from a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime. You’ll generally only need this to remove tags which were added by a previous call to put_tags().

This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.put_tags("images/front0_raw_000734.jpg", ["cloudy", "rainy"]})
>>>
>>> # Actually this is just Seattle's aggressive mist, that's not really rainy...
>>> file_changeset_manager.remove_tags("images/front0_raw_000734.jpg", ["rainy"]})
Parameters:

  • relative_path (str)

  • tags (list[str])

set_description(relative_path, description)#

Sets the human-readable description of a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.set_description("images/front0_raw_000734.jpg", "This image was over-exposed")
Parameters:

  • relative_path (str)

  • description (Optional[str])

On this page