Skip to content

Execute remotely

Every manager on AthenaClient is lazy and cached — the first access builds it; subsequent accesses share the same httpx connection pool.

from athena_sdk import AthenaClient

client = AthenaClient()                         # reads NEXUS_API_URL/KEY from env
run = client.executions.execute(workflow_id)
status = client.executions.get_status(workflow_id, run["execution_id"])

Streaming logs (SSE)

for event in client.executions.logs_stream(workflow_id, execution_id):
    print(event)

The iterator bypasses the generated client and streams events via httpx.stream() — upstream pressure naturally flows back to the server.

Error handling

from athena_sdk import AthenaSDKError, RemoteAPIError, ValidationError

try:
    client.workflows.create(name="")
except ValidationError as e:     # 422 from backend — typed
    for err in e.errors:
        print(err["loc"], err["msg"])
except RemoteAPIError as e:      # every non-2xx
    print(e)
except AthenaSDKError as e:      # catch-all for SDK failures
    print(e)