Advanced usage¶

Structured errors¶

All library exceptions derive from HubuumError. HTTP status codes are mapped to useful subclasses:

Other failed HTTP responses raise APIError. Network and TLS failures raise TransportError; successful responses that violate a typed model raise DecodeError.

from hubuum_client import NotFoundError

try:
    item = client.classes.get(42)
except NotFoundError as error:
    print(error.status_code, error.message, error.request_id)

Exceptions retain the method, query-free URL, status, API error code, message, and request ID where available. Bearer tokens, secret-bearing request headers and bodies, and values under sensitive query parameter names are redacted from their diagnostics.

RateLimitError.retry_after exposes a valid Retry-After delay as non-negative seconds. Both integer delay values and HTTP dates are supported; the attribute is None when the header is absent or malformed.

Complete OpenAPI operation surface¶

The Hubuum v0.0.16 OpenAPI contract contains 220 operations. Every operation is registered by its exact operationId, HTTP method, path template, request media types, and authentication policy:

from hubuum_client import OpenAPIOptions

result = client.openapi.call(
    "getApiV1Search",
    options=OpenAPIOptions(params={"q": "server", "limit_per_kind": 10}),
)

Path parameters are encoded as opaque segments, and missing or unexpected parameters fail before a request:

task = client.openapi.call(
    "postApiV1Imports",
    json=import_payload,
    options=OpenAPIOptions(headers={"Idempotency-Key": import_key}),
)

status = client.openapi.call(
    "getApiV1ImportsByTaskId",
    options=OpenAPIOptions(path_params={"task_id": task_id}),
)

Idempotency-Key values are validated before I/O and must contain between 1 and 255 bytes, matching v0.0.16 task-submission endpoints.

When an operation accepts multiple request representations, select one through content_type. Principal settings support JSON Merge Patch by default and RFC 6902 JSON Patch explicitly:

patched = client.openapi.call(
    "patchApiV1IamMeSettings",
    json=[{"op": "replace", "path": "/theme", "value": "dark"}],
    options=OpenAPIOptions(content_type="application/json-patch+json"),
)

call() returns a JSON value for JSON responses, str for a negotiated text response, bytes for any other media type, and None for an empty success. For example, rendered exports are not forced through a JSON decoder:

csv_text = client.openapi.call(
    "getApiV1ExportsByTaskIdOutput",
    options=OpenAPIOptions(
        path_params={"task_id": task_id},
        accept="text/csv",
    ),
)

The unified search stream is consumed incrementally:

with client.openapi.stream(
    "getApiV1SearchStream",
    options=OpenAPIOptions(params={"q": "server"}),
) as response:
    for line in response.iter_lines():
        process_sse_line(line)

Use async with and async for for the asynchronous client. The operation manifest is compared with the immutable server OpenAPI document in CI, including request and successful-response media types; all 220 operations must match exactly.

Structured search¶

Use postApiV1Search for the version 1 structured-search DSL. The same JSON envelope can select collections, classes, objects, audit events, users, groups, or service accounts. Object searches can select an exact class and combine field predicates with boolean and related-object predicates:

search = {
    "version": 1,
    "target": {"kind": "object", "class": {"name": "Servers"}},
    "filter": {
        "op": "field",
        "predicate": {"field": "name", "operator": "equals", "value": "web-01"},
    },
    "include_total": True,
    "limit": 25,
}
page = client.openapi.call("postApiV1Search", json=search)

with client.openapi.stream("postApiV1SearchStream", json=search) as response:
    for line in response.iter_lines():
        process_sse_line(line)

The JSON envelope returns tagged results, an optional total, and a next cursor. Pass next unchanged as the next request's cursor, preserving the search and authentication context. The SSE form emits started, result, and terminal done events; an error event signals failure after streaming begins. The done event carries cursor metadata. Use await client.openapi.call(...) and async with client.openapi.stream(..., json=search) in asynchronous code. GET search streams accept no body; POST search streams require one. See the server search reference for field, sort, and predicate limits.

Queued full restores¶

Hubuum v0.0.16 restore confirmation returns 202 Accepted when queued, before the separate administrator restore executor finishes. Use fresh credential approval with postApiV1RestoresByRestoreIdConfirm to confirm a validated stage, then poll getApiV1RestoresByRestoreIdStatus with the capability returned when staging:

status = client.openapi.call(
    "getApiV1RestoresByRestoreIdStatus",
    options=OpenAPIOptions(
        path_params={"restore_id": restore_id},
        headers={"X-Hubuum-Restore-Capability": restore_capability},
    ),
)

Repeat with an application-defined deadline and polling interval until status is succeeded or failed. This route sends no bearer token and works with a client that has no token; the capability authorizes the read after restored state replaces existing tokens. Keep the capability private. It is redacted from client error diagnostics and option representations. Full restore stages are separate from task IDs and cannot use client.tasks.wait().

Scoped tokens¶

Hubuum v0.0.16 nests token boundaries under one scope field. Omit scope for an unscoped token; within a scope, permissions and collection/class/object resources are independent dimensions:

from getpass import getpass

from hubuum_client import (
    NewTokenRequest,
    CreateTokenOperation,
    CredentialApprovalRequest,
    Permission,
    TokenResourceKind,
    TokenResourceScope,
    TokenScope,
)

request = NewTokenRequest(
    name="inventory-reader",
    scope=TokenScope(
        permissions=(Permission.READ_COLLECTION, Permission.READ_CLASS),
        resources=(TokenResourceScope(kind=TokenResourceKind.COLLECTION, id=collection_id),),
    ),
)
approval = client.credential_approvals.create(
    CredentialApprovalRequest(
        password=getpass("Current password: "),
        operation=CreateTokenOperation(principal_id=principal_id, token=request),
    )
)
token = client.tokens.for_principal(principal_id).create(request, approval=approval)

The returned AccessToken has a redacted string representation and exposes the authoritative expiry as token.expires_at. When NewTokenRequest.expires_at is omitted, the server uses client.config().authentication.default_token_lifetime_hours. Token metadata from client.me().token and token-list services uses scope is None to identify an unscoped token; the removed v0.0.3 flat scopes, resource_scopes, and scoped fields are not sent.

Token lists default to active credentials and accept TokenListState.EXPIRED, REVOKED, or ALL. A principal-specific service can inspect retained tokens and renew an active or expired token without exposing its previous secret:

from hubuum_client import RenewTokenOperation, RenewTokenRequest, TokenListState

tokens = client.tokens.for_principal(principal_id)
retained = tokens.list(state=TokenListState.ALL)
metadata = tokens.get(token_id)
renewal = RenewTokenRequest()
approval = client.credential_approvals.create(
    CredentialApprovalRequest(
        password=getpass("Current password: "),
        operation=RenewTokenOperation(principal_id=principal_id, token_id=token_id, token=renewal),
    )
)
replacement = tokens.renew(token_id, renewal, approval=approval)

Object aggregate measures¶

Aggregate endpoints accept up to three repeated group_by dimensions and four repeated aggregate measures. Each measure uses operation:field syntax:

rows = client.classes.by_name("Servers").object_aggregates(
    params=[
        ("group_by", "json_data.location,country"),
        ("aggregate", "sum:json_data.metrics,cpu"),
        ("aggregate", "average:json_data.metrics,memory_gib"),
    ]
)

for row in rows:
    for measure in row.measures:
        print(measure.operation, measure.value, measure.value_count, measure.skipped_count)

Global aggregation is available by sending one or more aggregate parameters without group_by.

Natural-key objects and JSON Patch¶

All class/object by-name aliases have first-class service paths:

from hubuum_client import ObjectDataPatchOperation, Query

hosts = client.classes.by_name("Hosts").objects
active = hosts.all(Query().data("status").equals("active"))
current = hosts.get("workstation.example")

updated = hosts.patch_data(
    current.name,
    [
        ObjectDataPatchOperation(
            op="add",
            path="/facts",
            value={"source": "ansible", "serial": "A"},
        )
    ],
)

client.classes.by_id(id).objects provides the corresponding numeric-ID interface. client.classes.by_name(name) additionally exposes class permissions, relations, graphs, aggregates, and its name-addressed object service. JSON Patch paths are relative to the raw data root. Operations are validated for their RFC 6902 member shape, explicit JSON null values are preserved, and the server applies the complete patch atomically with rename safety.

Typed imports, exports, and task events¶

Core import graphs use strict import-v2 request models, including the timestamps Hubuum v0.0.16 can restore. run() submits the task, waits with a bounded poller, and collects every per-entity result through guarded cursor pagination:

from datetime import datetime

from hubuum_client import (
    ImportCollectionInput,
    ImportGraph,
    ImportRequest,
    ImportWriteCondition,
    ImportWriteMode,
    RestoreTimestamps,
)

result = client.imports.run(
    ImportRequest(
        graph=ImportGraph(
            collections=(
                ImportCollectionInput(
                    ref_="inventory",
                    name="Inventory",
                    description="Imported inventory",
                    timestamps=RestoreTimestamps(
                        created_at=datetime(2024, 1, 1),
                        updated_at=datetime(2024, 1, 2),
                    ),
                    condition=ImportWriteCondition(mode=ImportWriteMode.CREATE_ONLY),
                ),
            )
        )
    ),
    idempotency_key="inventory-import-2024-01-02",
)
print(result.succeeded, result.failed)

The Python field ref_ is serialized as the contract's ref. Import graphs, object data, result details, and error strings are excluded from model representations. Integration-oriented import sections remain strict JSON objects so the full v0.0.16 graph can be submitted without representing secret configuration in diagnostic output. Core resources can use create_only, unconditional overwrite, or if_revision per-item write conditions.

Direct exports are similarly typed:

from hubuum_client import ExportRequest, ExportScope, ExportScopeKind

output = client.exports.run(
    ExportRequest(
        scope=ExportScope(
            kind=ExportScopeKind.OBJECTS_IN_CLASS,
            class_id=class_id,
        )
    ),
    idempotency_key="servers-export-1",
)

JSON output is returned as ExportJsonResponse; text, HTML, and CSV are returned as RenderedExport. Use client.exports.output_stream(task_id) for large output. Completed export task details expose total_duration_ms, query_duration_ms, hydration_duration_ms, and render_duration_ms. Task history is available through client.tasks.events(), event_pages(), and all_events(). Every method has a matching async form.

Custom extension routes¶

request() remains the lower-level escape hatch for a server extension that is not part of the pinned v0.0.16 OpenAPI document:

from hubuum_client import RequestOptions

result = client.request(
    "GET",
    "/api/v1/custom-extension",
    options=RequestOptions(params={"scope": "example"}),
)

The path must begin with one slash and may not contain an origin, traversal segment, query string, or fragment. Pass query values and additional headers through RequestOptions, and request bodies through json. Use response_model=MyPydanticModel to validate a custom JSON response.

Task polling¶

Imports, exports, backups, and computed-field rebuilds return tasks. Once a task ID is known, wait for a terminal state with a bounded poller:

task = client.tasks.wait(task_id, timeout_seconds=300, poll_interval=0.5)
print(task.status, task.progress.processed_items)

The async equivalent uses the same timeout_seconds and poll_interval keywords and awaits without blocking the event loop. Both pollers reject invalid bounds and cap each sleep to the remaining timeout. Import/export run() helpers raise TaskUnsuccessfulError, whose diagnostics contain only the task ID and status rather than potentially sensitive task summaries.