atlas-asset-client 0.2.1

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()})

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, 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.

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)

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.