Python API

SyncClient is the supported Python interface for the Sync HTTP API. It sends typed requests to the service and returns typed resources. It does not open a product store, load an adapter, resolve execution credentials, or run the execution core in the caller process.

The base infrahub-sync installation includes the client and HTTPX. FastAPI, Uvicorn, Prefect, Psycopg, and Boto3 remain optional service dependencies.

Connect

Construct a client with an absolute HTTP or HTTPS service URL, a bearer token, and a positive request timeout. TLS verification is enabled and redirects are refused.

HTTP provides no transport encryption for bearer credentials. Use HTTP only when the service connection stays inside a network you trust; use HTTPS across networks you do not trust.

from infrahub_sync.client import SyncClient

client = SyncClient("https://sync.example.com", "replace-with-a-token", timeout=30)

SyncClient.from_environment() reads INFRAHUB_SYNC_API_URL and INFRAHUB_SYNC_API_TOKEN:

from infrahub_sync.client import SyncClient

with SyncClient.from_environment(timeout=30) as client:
    version = client.get_version()
    status = client.get_status()

The first protected operation checks /version before sending the bearer token. The client requires the server to declare v3-unstable and caches a successful compatibility check for that client instance.

Register and inspect configurations

Configuration mutations require a reason and an idempotency key. A retry after an uncertain transport result must reuse the same key.

from infrahub_sync.client import ConfigMutationRequest, SyncClient

package = {
    "format_version": 1,
    "configuration": {
        "name": "inventory",
        "source": {"name": "netbox", "settings": {}},
        "destination": {"name": "infrahub", "settings": {}},
    },
}

with SyncClient.from_environment() as client:
    registered = client.register_config(
        ConfigMutationRequest(package=package, reason="Register inventory sync"),
        idempotency_key="register-inventory-20260830",
    )
    config_id = registered.configuration.config_id
    registry_version = registered.version.registry_version
    report = client.validate_config(config_id, registry_version, offset=0, limit=256)

The client also exposes create_config_version, list_configs, get_config, list_config_versions, and get_config_version. Configuration list methods return the complete server response and do not expose pagination arguments. Validation exposes the server's finding page fields and preserves finding order.

Create and inspect runs

Create a plan from a registered configuration version:

from infrahub_sync.client import CreateRunRequest, SyncClient

with SyncClient.from_environment() as client:
    accepted = client.plan(
        CreateRunRequest(
            operation="plan",
            config_id=config_id,
            registry_version=registry_version,
            branch="main",
            confirm_writes=False,
            reason="Review inventory changes",
        ),
        idempotency_key="plan-inventory-20260830",
    )
    finished = client.wait_for_run(accepted, timeout=1800, poll_interval=2)
    saved_plan = client.get_plan(finished.run.run_id)

A confirmed composed sync uses operation="sync" and confirm_writes=True. The client also exposes get_run, get_results, list_artifacts, get_artifact, verify_run, apply_run, and cancel_run. The short method names sync, verify, and apply call the same HTTP operations.

apply_run sends the expected checksum from the reviewed PlanResource. It does not read or reconstruct plan bytes locally. get_artifact verifies the response Digest header against the received bytes before returning ArtifactContent.

Wait behavior

wait_for_run follows the final orchestration entry in the accepted mutation response by its flow_run_id. Earlier completed plan or verification executions do not make a later apply appear complete. Unknown non-terminal live states continue polling until the bounded deadline.

The durable terminal outcome decides the result:

• completed/succeeded returns the latest RunResource;
• failed, cancelled, abandoned, or ambiguous outcomes raise RunTerminalError;
• reaching the deadline raises RunWaitTimeoutError with the run ID and last observed product and execution state.

A local KeyboardInterrupt stops the wait and crosses the Python boundary unchanged. It does not cancel the remote run.

Errors

Catch SyncClientError for all client-owned failures. More specific types are available for input, compatibility, transport, protocol, API refusal, configuration API refusal, wait timeout, and terminal run failures.

from infrahub_sync.client import ConfigsAPIError, SyncClientError

try:
    configurations = client.list_configs()
except ConfigsAPIError as exc:
    print(exc.status, exc.code, exc.family, exc.reason)
except SyncClientError:
    raise

Exceptions contain fixed client-owned text. Valid API refusals retain only their documented machine fields. Raw response bodies, response headers, bearer values, and HTTPX exception chains do not cross the client boundary.