cloud-bridge-client 0.0.2

Client for integration with Cloud Bridge

cloud-bridge-client

A lightweight, typed Python client for the Cloud Bridge API — a multi-cloud infrastructure orchestration service.

Cloud Bridge lets you describe an entire cluster (networks, firewalls, nodes, volumes, NAT gateways, outputs, …) in a single provider-agnostic YAML file and deploy it to AWS, GCP or OpenStack through one uniform HTTP API. This library is the Python front-end to that API: you pass credentials and a config file, and you get back structured results and typed exceptions.

from cloud_bridge_client import AwsClient, AWSCredentials

client = AwsClient(
    credentials=AWSCredentials(
        access_key_id="AKIA...",
        secret_access_key="...",
        region="eu-central-1",
        zone="eu-central-1a",
    ),
)

client.preview(config="cluster.yaml")   # dry-run
client.deploy(config="cluster.yaml")     # provision infrastructure

---

Table of contents

• Features
• Requirements
• Installation
• Quick start
• Authentication & credentials
• The cluster config file
• Operations reference
• Synchronous vs. asynchronous operations
• Destroy: the two-phase confirmation flow
• Error handling
• Full example
• License

---

Features

• One client per provider — AwsClient, GcpClient, OpenStackClient — with an identical method surface.
• Provider-agnostic config — the same cluster.yaml deploys to any supported cloud.
• Typed credentials — each provider has its own dataclass; no more juggling loose strings.
• Rich exception hierarchy — HTTP errors are mapped to specific, catchable exceptions (AuthenticationError, FlavorError, ValidationError, …) carrying the server's structured error payload.
• Flexible config input — pass a file path, raw YAML bytes, or a plain dict.
• Two-phase destroy — a built-in confirmation flow protects against accidental teardown.
• Zero-magic transport built on requests; no async runtime required.

Requirements

• Python 3.13+
• A running Cloud Bridge server that the client can reach over HTTP.
• Valid cloud-provider credentials for whichever provider you target.

Installation

pip install cloud-bridge-client

Dependencies (requests, pyyaml, python-dotenv) are installed automatically.

Quick start

Server URL. base_url is optional and defaults to http://localhost:8080. If your Cloud Bridge server lives elsewhere, pass base_url= to the provider client you are constructing (e.g. AwsClient(credentials=..., base_url="http://my-server:8080")). The timeout argument (seconds) is optional too and defaults to 600.

from cloud_bridge_client import AwsClient, AWSCredentials

client = AwsClient(
    credentials=AWSCredentials(
        access_key_id="AKIA...",
        secret_access_key="...",
        region="eu-central-1",
        zone="eu-central-1a",
    ),
)

# Is the server reachable?
assert client.ping()

# What can this server provision?
print(client.providers())          # -> ["aws", "gcp", "openstack", ...]

# Dry-run first — see what *would* change, touching nothing.
print(client.preview(config="cluster.yaml"))

# Provision. Returns a job_id — the work runs asynchronously on the server.
result = client.deploy(config="cluster.yaml")
job_id = result["data"]["job_id"]

# Inspect a running/finished stack.
print(client.status(cluster_name="swarm-cluster"))

Authentication & credentials

Every client is constructed with a credentials object for its provider. Under the hood the library encodes the credentials into a single X-API-Key header — you never build that header yourself.

Pick the client + credentials pair that matches your target cloud:

AWS

from cloud_bridge_client import AwsClient, AWSCredentials

client = AwsClient(
    credentials=AWSCredentials(
        access_key_id="AKIA...",
        secret_access_key="...",
        region="eu-central-1",   # default: "eu-central-1"
        zone="eu-central-1a",    # default: ""
    ),
)

GCP

from cloud_bridge_client import GcpClient, GCPCredentials

client = GcpClient(
    credentials=GCPCredentials(
        project="my-project",
        region="us-central1",
        zone="us-central1-a",
        service_account_email="[e-mail]",
        private_key="-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----\n",
    ),
)

OpenStack

from cloud_bridge_client import OpenStackClient, OpenStackCredentials

client = OpenStackClient(
    credentials=OpenStackCredentials(
        auth_url="https://my-openstack:5000/v3",
        application_credential_id="...",
        application_credential_secret="...",
        region_name="RegionOne",   # default: "RegionOne"
    ),
)

Constructor parameters (shared by every client):

Parameter	Type	Default	Meaning
credentials	provider credentials dataclass	required	Provider credentials.
base_url	str	http://localhost:8080	Base URL of the Cloud Bridge server.
timeout	int	600	Per-request timeout in seconds.

Hot-swapping credentials

You can rotate credentials on a live client without recreating it:

client.update_credentials(AWSCredentials(access_key_id="AKIA-NEW...", secret_access_key="..."))

Loading credentials from the environment

Credentials are plain dataclasses, so any config source works. A common pattern with python-dotenv (bundled as a dependency):

import os
from dotenv import load_dotenv
from cloud_bridge_client import AwsClient, AWSCredentials

load_dotenv()

client = AwsClient(
    credentials=AWSCredentials(
        access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
        secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
        region=os.environ["AWS_REGION"],
        zone=os.environ["AWS_AVAILABILITY_ZONE"],
    ),
)

The cluster config file

deploy and preview take a cluster config describing the infrastructure you want. It is the same provider-agnostic YAML the Cloud Bridge server understands — networks, firewalls, NAT gateways, keypairs, volumes, nodes and outputs. The server resolves provider-specific details (image names, flavors, sources) from per-provider keys inside the file.

You can supply the config three ways:

# 1. A path to a file on disk (str or pathlib.Path) — must exist.
client.deploy(config="infra/cluster.yaml")

# 2. Raw YAML bytes already in memory.
client.deploy(config=b"name: swarm-cluster\nnodes: ...")

# 3. A plain dict — serialised to YAML for you.
client.deploy(config={"name": "swarm-cluster", "nodes": [...]})

A minimal shape looks like:

name: swarm-cluster
description: Docker Swarm cluster on Fedora CoreOS

networks:
  - create: true
    name: swarm-internal-net
    cidr: "10.10.0.0/24"
    external_network: public

firewalls:
  - name: swarm-cluster-sg
    ingress:
      - protocol: tcp
        port: 22
        cidr: "0.0.0.0/0"

nodes:
  - name: swarm-manager
    flavor:
      name: t3.small
      vcpus: 1
      ram_gb: 3
      architecture: x86_64
    image:
      name:
        aws: fedora-coreos-44...x86_64
        gcp: fedora-coreos-stable
        openstack: Fedora-CoreOS-stable
      user_data_format: ignition
    public_ip: true

outputs:
  - name: manager_public_ip
    from: swarm-manager.public_ip

The config schema is owned and validated by the Cloud Bridge server, not by this client. See the server documentation (and the universal_cluster.yaml example shipped in this repository) for the full set of supported keys.

Automatic flavor selection

If your config does not pin a flavor for a node, pass auto_select=True and the server will pick one that satisfies the requested vcpus / ram_gb / architecture:

client.preview(config="cluster.yaml", auto_select=True)
client.deploy(config="cluster.yaml", auto_select=True)

Operations reference

Every client exposes the same methods:

Method	Kind	Description
ping()	sync	Returns True if the server is reachable (never raises).
providers()	sync	Returns a list[str] of enabled provider identifiers.
preview(config, *, auto_select=False, request_id=None)	sync	Dry-run of a deploy — shows planned changes without touching real infra.
deploy(config, *, auto_select=False, request_id=None)	async job	Provisions the stack. Returns a job_id.
status(cluster_name)	sync	Current outputs + last-update summary for a stack.
volumes(cluster_name)	sync	What will happen to each volume when the stack is destroyed.
preview_destroy(cluster_name)	sync	Dry-run of a destroy — lists resources that would be deleted.
destroy(cluster_name, *, ...)	async job	Tears down a stack. Two-phase (see below).
cancel(cluster_name)	sync	Cancels an in-progress update / clears a stale lock.
update_credentials(credentials)	—	Hot-swaps credentials on the live client.

Method signatures

def deploy(self, config, *, auto_select=False, request_id=None) -> Any
def preview(self, config, *, auto_select=False, request_id=None) -> Any
def preview_destroy(self, cluster_name: str) -> Any
def cancel(self, cluster_name: str) -> Any
def destroy(self, cluster_name, *, stateless_resources=False,
            stateful_resources=False, config_file=False,
            confirm_code=None, request_id=None) -> Any
def providers(self) -> list[str]
def volumes(self, cluster_name: str) -> Any
def status(self, cluster_name: str) -> Any
def ping(self) -> bool
def update_credentials(self, credentials) -> None

Most methods return the server's parsed JSON envelope, shaped like:

{ "success": true, "data": { ... } }

providers() and ping() are the exceptions — they unwrap the useful value for you (a list and a bool respectively).

The optional request_id (a UUID v4) is echoed back through the server for request tracing; if you omit it, the server generates one.

Synchronous vs. asynchronous operations

The Cloud Bridge server runs long-lived infrastructure work on a background queue. This has a concrete consequence for the client:

• deploy and destroy are asynchronous. They return quickly with a job_id; the actual provisioning/teardown continues on the server afterwards.

  result = client.deploy(config="cluster.yaml")
  job_id = result["data"]["job_id"]
  # -> {"data": {"job_id": "...", "operation": "deploy",
  #              "provider": "aws", "cluster_name": "swarm-cluster", "flavor": ...}}
  

• Everything else is synchronous — preview, preview_destroy, status, volumes, providers, cancel all complete within the single HTTP call.

To follow the progress of a deploy/destroy job, poll status(cluster_name=...). Because a deploy can take a long time, the default timeout is a generous 600 seconds — tune it to your environment.

Destroy: the two-phase confirmation flow

Because destroying infrastructure is irreversible, destroy() uses a two-step handshake.

Phase 1 — request confirmation. Call with at least one of stateless_resources, stateful_resources, or config_file set to True. Nothing is deleted yet; the server returns a short-lived confirmation code and a warning describing exactly what will be removed.

resp = client.destroy(
    cluster_name="swarm-cluster",
    stateless_resources=True,   # compute, networks, ...
    config_file=True,           # also remove the Pulumi config file
)
data = resp["data"]
print(data["warning"])          # human-readable warning
code = data["code"]             # e.g. "a1b2c3d4" — expires after a short TTL
print(data["volumes"])          # volumes that would be affected

Phase 2 — confirm. Call again with confirm_code set to the code from phase 1. This enqueues the actual teardown and returns a job_id.

resp = client.destroy(cluster_name="swarm-cluster", confirm_code=code)
job_id = resp["data"]["job_id"]

Flag meanings:

Flag	Deletes
stateless_resources=True	Compute, networks, and other stateless resources.
stateful_resources=True	Volumes, databases (requires stateless_resources=True).
config_file=True	The server-side Pulumi config file for the stack.

The confirmation code expires. If it lapses, phase 2 fails and you must repeat phase 1.

Error handling

All exceptions inherit from CloudBridgeClientError, so you can catch broadly or narrowly. Non-2xx HTTP responses are mapped to specific subclasses of ApiError, each carrying the server's status_code, message, and structured data payload.

from cloud_bridge_client import (
    AuthenticationError,
    FlavorError,
    ValidationError,
    StackOperationError,
    ServerConnectionError,
    ApiError,
    CloudBridgeClientError,
)

try:
    client.deploy(config="cluster.yaml", auto_select=True)
except FlavorError as e:
    # 400 — the requested flavor is invalid; server returns candidates in e.data
    print("Available flavors per node:", e.data)
except ValidationError as e:
    # 422 — payload rejected; e.data is {field: [messages]}
    print("Validation failed:", e.data)
except AuthenticationError:
    # 401 / 403 — bad credentials
    print("Check your credentials.")
except ServerConnectionError:
    # Could not reach the Cloud Bridge server at all
    print("Server unreachable.")
except ApiError as e:
    # Any other non-2xx response
    print(f"HTTP {e.status_code}: {e.message}")

Exception hierarchy

CloudBridgeClientError                 # base for everything
├── ServerConnectionError              # cannot reach the Cloud Bridge server
├── ConfigError                        # local config could not be read/parsed
├── ProviderMismatchError              # wrong client for the credential type
└── ApiError                           # any non-2xx HTTP response (has .status_code, .message, .data)
    ├── AuthenticationError            # 401 / 403
    ├── ProviderNotFoundError          # 404 (region/zone/stack not found)
    ├── ProviderConnectionError        # 400 — server could not reach the cloud provider
    ├── ValidationError                # 422 — request payload rejected
    ├── StackOperationError            # 409 — deploy/destroy/preview conflict
    ├── FlavorError                    # 400 — flavor validation failed (candidates in .data)
    ├── ImageError                     # 400 — image validation failed (candidates in .data)
    ├── VolumeError                    # 400 — volume-type validation failed (candidates in .data)
    └── NetworkError                   # 400 — network validation failed (detail in .data)

Each ApiError renders nicely when printed — str(e) includes the status code, message, and a pretty-printed data payload when present, which makes FlavorError / ImageError / VolumeError immediately actionable.

Full example

An interactive, env-driven example lives in main.py. The essence:

import os
from pprint import pprint

from dotenv import load_dotenv
from cloud_bridge_client import AwsClient, AWSCredentials

load_dotenv()

client = AwsClient(
    credentials=AWSCredentials(
        access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
        secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
        region=os.environ["AWS_REGION"],
        zone=os.environ["AWS_AVAILABILITY_ZONE"],
    ),
)

CONFIG = "universal_cluster.yaml"
CLUSTER = "swarm-cluster"

pprint(client.preview(config=CONFIG, auto_select=True))   # dry-run
pprint(client.deploy(config=CONFIG))                      # provision
pprint(client.status(cluster_name=CLUSTER))               # follow up

# Two-phase teardown
resp = client.destroy(cluster_name=CLUSTER, stateless_resources=True, config_file=True)
code = resp["data"]["code"]
if input(f"Confirm destroy? code={code} (y/n) ") == "y":
    pprint(client.destroy(cluster_name=CLUSTER, confirm_code=code))

A matching .env.example with the expected variable names is included in the repository.

License

MIT © Aurora Resourcing.