roboto.action_runtime
Submodules
Package Contents
ActionInput
Resolved references to input data an Action was given to operate on.
To use, access via get_input().
Usage
From within an Action, list files passed as input and check their size:
context = InvocationContext.from_env()
action_input = context.get_input()
for file, local_path in action_input.files:
print(f"{file.file_id} is {local_path.stat().st_size} bytes")Attributes
ActionInput.files
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
ActionInput.from_record()
Create an ActionInput instance from its serialized representation.
Parameters
record ActionInputRecordroboto_client roboto.Return type
ActionInput.get_topics_by_name()
Return any topics in this ActionInput that have the provided name.
Parameters
topic_name strTopic 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.
Attributes
ActionInput.sessions
Sessions passed as input data to an action invocation.
ActionInput.topics
Topics passed as input data to an action invocation.
ActionInputResolver
Resolves the action invocation input spec to concrete Roboto entities.
The entities are packaged together in an ActionInput instance, which is available to action code via ActionRuntime.
Parameters
file_service roboto.ActionInputResolver.from_env()
Parameters
roboto_client Optional[roboto.roboto_search Optional[roboto.Return type
ActionInputResolver.resolve_input_spec()
Resolve the input spec’s file, session, and topic selectors to the entities they match.
When the spec asks for files, sessions, or topics and none match, that category comes back empty with a logged warning rather than an error.
Parameters
input_spec roboto.Input specification containing data selectors. See InvocationInput for more detail.
download boolIf True, download all resolved files to local disk. Defaults to False.
download_path Optional[pathlib.Directory path where files should be downloaded. If not provided and download=True, a temporary directory will be created. Ignored if download=False.
Returns
ActionInputRecord containing:
- files: List of (FileRecord, Optional[Path]) tuples. Path is None if download=False, otherwise contains the local path where the file was downloaded.
- sessions: List of SessionRecord instances.
- topics: List of TopicRecord instances.
Usage
Resolve files using a RoboQL query without downloading:
input_spec = InvocationInput.file_query('dataset_id = "ds_abc123" AND path LIKE "%.mcap"')
result = resolver.resolve_input_spec(input_spec)
# result.files contains (FileRecord, None) tuples
# result.topics is emptyResolve and download files to a specific directory:
input_spec = InvocationInput.file_query('dataset_id = "ds_abc123" AND path LIKE "%.mcap"')
result = resolver.resolve_input_spec(input_spec, download=True, download_path=Path("/tmp/data"))
# result.files contains (FileRecord, Path) tuples with local pathsResolve both files and topics:
input_spec = InvocationInput(
files=FileSelector(query='dataset_id = "ds_abc123" AND path LIKE "%.mcap"'),
topics=DataSelector(names=["battery_status", "gps"]),
)
result = resolver.resolve_input_spec(input_spec)
# result.files contains file records
# result.topics contains topic recordsActionRuntime
Bases: InvocationContext
Deprecated. Use InvocationContext instead.
ActionRuntimeException
Bases: Exception
Base class for all exceptions raised by the action_runtime submodule.
ExitCode
Bases: enum.IntEnum
Defined exit codes used by the action runtime. Exception codes are adapted from /usr/include/sysexits.h
Attributes
ExitCode.ConfigurationError
From /usr/include/sysexits.h: > EX_CONFIG 78 /* configuration error */
ExitCode.DataError
From /usr/include/sysexits.h:
> EX_DATAERR – The input data was incorrect in some way. > This should only be used for user’s data & not system > files.
We use this in our own ingestion actions to signify “the ingestion action did the right thing, but the input file was the wrong format, corrupted, a 0-byte file, etc.”
ExitCode.InternalError
From /usr/include/sysexits.h: > EX_SOFTWARE – An internal software error has been detected. > This should be limited to non-operating system related > errors as possible.
ExitCode.Success
ExitCode.UsageError
From /usr/include/sysexits.h: > EX_USAGE – The command was used incorrectly, e.g., with > the wrong number of arguments, a bad flag, a bad > syntax in a parameter, or whatever.
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.InvocationContext.
FilesChangesetFileManager.put_fields()
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.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.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 strmetadata dict[str, Any]FilesChangesetFileManager.put_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.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.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 strtags list[str]FilesChangesetFileManager.remove_fields()
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.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.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 strkeys list[str]FilesChangesetFileManager.remove_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.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.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 strtags list[str]FilesChangesetFileManager.set_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.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.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 strdescription Optional[str]InvocationContext
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 InvocationContext
context = InvocationContext.from_env()…which will inspect environment variables to initialize the InvocationContext.
If you want to test a script using InvocationContext 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, initialize InvocationContext directly:
import pathlib
from roboto import InvocationContext
context = InvocationContext(
dataset_id="ds_XXXXXXXXXXXX",
input_dir=pathlib.Path("/path/to/tmp/input/dir"),
invocation_id="iv_XXXXXXXXXXXX",
org_id="og_XXXXXXXXXXXX",
output_dir=pathlib.Path("/path/to/tmp/output/dir"),
)Parameters
dataset_id strinput_dir pathlib.invocation_id strorg_id stroutput_dir pathlib.input_data_manifest_file Optional[pathlib.parameters_file Optional[pathlib.secrets_file Optional[pathlib.roboto_client Optional[roboto.dry_run boollog_level Optional[str]Properties
InvocationContext.dataset
A Dataset instance for the dataset whose data this action is operating on, if any.
This resource will be lazily initialized the first time it is accessed. After the first call, the dataset will be cached.
This is particularly useful for adding tags or metadata to a dataset at runtime.
Raises
If the dataset does not exist.
If the dataset is not specified (e.g., when running locally, using scheduled triggers, or invoking via CLI with query-based input data).
Return type
Usage
Add tags/metadata to the dataset:
context.dataset.put_tags(["tagged_by_action"])
context.dataset.put_metadata({"voltage_spikes_seen": 693})InvocationContext.dataset_id
The ID of the dataset whose data this action is operating on.
InvocationContext.file_changeset_manager
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 InvocationContext
context = InvocationContext.from_env()
my_output_file = context.output_dir / "my_output_file.txt"
my_output_file.write_text("Hello World")
context.file_changeset_manager.put_tags(my_output_file.name, ["tagged_by_action"])
context.file_changeset_manager.put_fields(
my_output_file.name, {"roboto_proficiency": "extreme - I can annotate output files!"}
)This only works for files that have not yet been uploaded to Roboto. To tag existing files, you should instead use:
from roboto import InvocationContext
context = InvocationContext.from_env()
existing_file = context.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.
InvocationContext.from_env()
Initialize an InvocationContext from values in environment variables. Will throw an exception if any required environment variables are not available.
All required environment variables will be available at runtime when an action is running in Roboto’s remote execution environment.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()Return type
InvocationContext.get_input()
Instance of ActionInput containing resolved references to input data.
InvocationContext.get_optional_parameter()
Retrieve the value of the action parameter with the given name, defaulting to default_value if the parameter is not set.
Parameters
name strThe name of the parameter to retrieve.
default_value Optional[str]The value to return if the parameter is not set. Defaults to None.
Returns
The parameter value, or default_value if not set. If the value is a secret URI, returns the resolved secret value.
Usage
import roboto
context = roboto.InvocationContext.from_env()
context.get_optional_parameter("model_version", "latest")
# "latest"InvocationContext.get_parameter()
Gets the value of the action parameter with the given name, raising an ActionRuntimeException if the parameter is not set.
Parameters
name strReturn type
InvocationContext.get_secret_parameter()
Gets the value of the secret action parameter with the given name.
Parameters
name strReturn type
Properties
InvocationContext.input_dir
The directory where the action’s input files are located.
InvocationContext.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.
InvocationContext.invocation_id
The ID of the currently running action invocation.
InvocationContext.log_level
The log level for the action invocation.
Returns
The log level constant (e.g., logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR) if set, or logging.INFO if no log level was specified.
InvocationContext.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.
InvocationContext.org_id
The ID of the org which invoked the currently running action.
InvocationContext.output_dir
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.
InvocationContext.roboto_client
The RobotoClient instance used by this action runtime.
PrepareEnvException
Bases: ActionRuntimeException
Base class for all exceptions raised by the action_runtime submodule.
Parameters
reason strAttributes
PrepareEnvException.exit_code
PrepareEnvException.reason
prepare_invocation_input_data()
Parameters
requires_downloaded_inputs boolinput_data Optional[roboto.target_directory pathlib.download_directory pathlib.inputs_data_manifest_file pathlib.roboto_client roboto.roboto_search roboto.prepare_invocation_parameters()
Parameters
action_parameters collections.provided_parameter_values collections.parameters_values_file pathlib.secrets_file pathlib.org_id Optional[str]roboto_client roboto.prepare_metadata_changeset_manifest()
Parameters
dataset_metadata_changeset_path pathlib.