SDK & CLI#

Tip

Our Python SDK is open source! Check out github.com/roboto-ai/roboto-python-sdk.

Overview#

The Python SDK and CLI both provide programmatic access to Roboto.

The CLI is convenient for quickly creating datasets, uploading or downloading files, and running actions. The Python SDK supports the full range of Roboto platform features and is particularly useful for data analysis and integration with other tools.

See reference documentation for the Python SDK and the CLI.

Install#

Refer to the latest installation instructions available on GitHub.

CLI#

With the Python SDK or standalone CLI installed, you can use roboto on the command line.

The following example creates a new dataset and uploads a file to it:

$ roboto datasets create --tag boston
{
  "administrator": "Roboto",
  "created": "2024-09-25T22:22:48.271387Z",
  "created_by": "benji@roboto.ai",
  "dataset_id": "ds_9ggdi910gntp",
  ...
  "tags": [
    "boston",
  ]
}

$ roboto datasets upload-files -d ds_9ggdi910gntp -p scene57.bag
100.0%|█████████████████████████ | 58.9M/58.9M | 2.62MB/s | 00:23 | Src: 1 file

Python SDK#

With the Python SDK installed, import roboto into your Python runtime.

The following example accesses topic data for an ingested ROS bag file:

from roboto import Dataset

ds = Dataset.from_id("ds_9ggdi910gntp")
bag = ds.get_file_by_path("scene57.bag")
steering_topic = bag.get_topic("/vehicle_monitor/steering")

steering_data = steering_topic.get_data(
    start_time="1714513576", # "<sec>.<nsec>" since epoch
    end_time="1714513590",
)

You can also create events:

from roboto import Event

Event.create(
  start_time="1714513580", # "<sec>.<nsec>" since epoch
  end_time="1714513590",
  name="Fast Turn",
  topic_ids=[steering_topic.topic_id]
)

Or search for logs matching metadata and statistics with RoboQL:

from roboto import query, RobotoSearch
roboto_search = RobotoSearch(query.client.QueryClient())

query = '''
dataset.tags CONTAINS 'boston' AND
topics[0].msgpaths[/vehicle_monitor/vehicle_speed.data].max > 20
'''

results = roboto_search.find_files(query)

Refer to the collection of notebooks for complete examples, and the SDK reference.

Handling Eventual Consistency#

Some Roboto API operations use eventually consistent reads. Query results may not immediately reflect recent writes; replication lag is typically under one second.

Affected operations include RobotoSearch.find_files(), RobotoSearch.find_datasets(), and other search methods, as well as fetching results or counts from submitted queries. This matters most when you create or update data and query for it immediately — for example, in automated tests or workflows with read-after-write dependencies.

Retry Pattern#

To ensure recently written data appears in query results, use a retry with a brief delay:

import time
from roboto import Dataset, RobotoSearch, query

# Create a dataset
ds = Dataset.create(tags=["my-new-tag"])

# Search for it with retry logic
roboto_search = RobotoSearch(query.client.QueryClient())
max_attempts = 5
delay_seconds = 1

for attempt in range(max_attempts):
    results = roboto_search.find_datasets('tags CONTAINS "my-new-tag"')

    if any(r.dataset_id == ds.dataset_id for r in results):
        print(f"Found dataset after {attempt + 1} attempt(s)")
        break

    if attempt < max_attempts - 1:
        time.sleep(delay_seconds)
else:
    print("Dataset not found after retries")

For most use cases, a single retry after one second is sufficient.

Best Practices#

  • Do not poll continuously without delays.

  • Most data replicates within one to two seconds; set timeouts accordingly.

  • Structure workflows to avoid immediate read-after-write dependencies where possible.

  • If you need guaranteed consistency, use direct record lookups (e.g., Dataset.from_id()) instead of search queries.

For details on which API endpoints use eventual consistency, see the REST API documentation.

On this page