atlas-asset-client 0.2.2

Async HTTP client for Atlas Command.

Atlas Command HTTP Client (Python)

atlas-asset-client is a lightweight async wrapper around the Atlas Command REST API. It replaces the retired WebSocket helpers and provides strongly-typed convenience methods for working with entities, tasks, objects, and query endpoints via HTTP.

Installation

pip install atlas-asset-client

During local development inside this repository:

pip install -e Atlas_Command/connection_packages/atlas_asset_ws_client

Quickstart

import asyncio
from atlas_asset_ws_client import AtlasCommandHttpClient

async def main() -> None:
    async with AtlasCommandHttpClient("http://localhost:8000") as client:
        entity = await client.create_entity(
            entity_id="asset-1",
            entity_type="asset",
            alias="Demo Asset",
            components={"telemetry": {"latitude": 40.7, "longitude": -74.0}},
        )
        print("Created entity:", entity["entity_id"])

        tasks = await client.list_tasks(limit=10)
        print("Existing tasks:", [t["task_id"] for t in tasks])

        snapshot = await client.get_full_dataset()
        print("Snapshot counts:", {k: len(v) for k, v in snapshot.items()})

        with open("mission_video.mp4", "rb") as video:
            stored = await client.create_object(
                file=video,
                object_id="mission-video-1",
                usage_hint="mission_video",
                referenced_by=[{"entity_id": "asset-1"}],
            )
            print("Uploaded object:", stored["object_id"])

asyncio.run(main())

Features

• Uses httpx.AsyncClient under the hood with pluggable transport/timeouts.
• Convenience methods for every public endpoint:
  • list_entities, get_entity, create_entity, update_entity, delete_entity, get_entity_by_alias, update_entity_telemetry
  • list_tasks, create_task, get_task, update_task, delete_task, get_tasks_by_entity, start_task, complete_task, fail_task
• list_objects, create_object (uploads a file via /objects/upload), get_object,
• update_object, delete_object,
• get_objects_by_entity, get_objects_by_task, find_orphaned_objects,
• add_object_reference, remove_object_reference, get_object_references,
• validate_object_references, cleanup_object_references
  • get_changed_since, get_full_dataset
• Optional bearer token support via the token= constructor parameter.
• Context manager support (async with client:) to manage connection lifecycle.

Field reference

Client configuration

• AtlasCommandHttpClient(base_url, *, token=None, timeout=10.0, transport=None) – requires base_url, optional token, timeout, and transport.

Entities

• list_entities(*, limit=100, offset=0) – optional pagination parameters based on defaults.
• get_entity(entity_id) – requires entity_id.
• get_entity_by_alias(alias) – requires alias.
• create_entity(*, entity_id, entity_type, alias, components=None) – requires entity_id, entity_type, and alias; components are optional.
• update_entity(entity_id, *, components=None) – requires entity_id, optional component patch.
• delete_entity(entity_id) – requires entity_id.
• update_entity_telemetry(entity_id, *, latitude=None, longitude=None, altitude_m=None, speed_m_s=None, heading_deg=None) – requires entity_id; telemetry values are optional and only set when provided.

Tasks

• list_tasks(*, status=None, limit=25) – optional status and page size.
• get_task(task_id) – requires task_id.
• create_task(*, task_id, components=None) – requires task_id; components optional.
• update_task(task_id, *, components=None) – requires task_id; components optional.
• delete_task(task_id) – requires task_id.
• get_tasks_by_entity(entity_id, *, status=None, limit=25) – requires entity_id; filters optional.
• start_task(task_id) / complete_task(task_id) – each requires task_id.
• fail_task(task_id, *, error_message=None, error_details=None) – requires task_id; error info optional.

Objects

• list_objects(*, content_type=None, limit=100, offset=0) – optional filters.
• get_object(object_id) – requires object_id.
• create_object(file, *, object_id, usage_hint=None, referenced_by=None) – requires file data and object_id; usage_hint and referenced_by optional.
• update_object(object_id, *, usage_hints=None, referenced_by=None) – requires object_id; metadata optional.
• delete_object(object_id) – requires object_id.
• get_objects_by_entity(entity_id, *, limit=50) – requires entity_id, optional pagination.
• get_objects_by_task(task_id, *, limit=50) – requires task_id, optional pagination.
• add_object_reference(object_id, *, entity_id=None, task_id=None) / remove_object_reference(...) – require object_id; provide either entity_id or task_id to target the reference.
• find_orphaned_objects(*, limit=100) – optional limit.
• get_object_references(object_id) / validate_object_references(object_id) / cleanup_object_references(object_id) – each requires object_id.

Queries

• get_changed_since(since, *, limit_per_type=None) – requires since; optional per-type limit.
• get_full_dataset(*, entity_limit=None, task_limit=None, object_limit=None) – filters are optional.

Configuration

client = AtlasCommandHttpClient(
    "https://atlas.example.com",
    token="my-api-token",
    timeout=30.0,
)

You can also pass a custom httpx transport for testing:

transport = httpx.MockTransport(my_handler)
client = AtlasCommandHttpClient("http://testserver", transport=transport)

Breaking changes

• create_entity now requires alias and no longer accepts published_at, updated_at, or extra.
• create_task / update_task drop status/extra; lifecycle helpers no longer accept started_by or result.
• create_object is the only object-creation helper and always uploads a file via /objects/upload; storage metadata and sizing is server-managed.

Testing

Run the suite with:

pip install -e .[dev]
pytest

The tests use httpx.MockTransport so they do not require a running Atlas Command instance.