foreman-client

A generic abstraction client library for the Red Hat Satellite / Foreman API, written in Python using requests.

Rather than hand-coding a Python method for every one of Foreman/Katello's hundreds of API endpoints, this library provides a single dynamic client that maps Python attribute access directly onto REST paths. Any endpoint under /api/v2 or /katello/api/v2 is reachable without needing per-resource code added to the library.

Features

• Full API coverage via a dynamic resource wrapper — no need to wait for a method to be added for a given endpoint.
• Sync HTTP client built on requests.Session.
• Two auth modes: username/password (HTTP Basic) or API token (Bearer).
• Automatic pagination — .list() walks all pages and returns every result as a single list.
• Katello namespace — client.katello.* automatically targets /katello/api/v2 instead of /api/v2.
• Custom/non-CRUD actions — call arbitrary action endpoints (e.g. power, errata/apply) via .action().
• Typed exception hierarchy — distinct exceptions for auth, not-found, validation, and generic API errors.
• Client-side field projection — trim returned records down to specific keys with fields=.
• SSL verification toggle — defaults to verified, can be disabled per-client.

Installation

pip install .

or, for local development:

pip install -e .

Requires Python >= 3.8 and requests >= 2.25.0.

Quick start

from foreman_client import ForemanClient

client = ForemanClient(
    "https://satellite.example.com",
    username="admin",
    password="secret",
    verify_ssl=True,   # set False for self-signed certs (not recommended for prod)
)

hosts = client.hosts.list()

Authenticating with an API token instead

client = ForemanClient(
    "https://satellite.example.com",
    api_token="your-api-token-here",
)

You must supply either api_token, or both username and password. Supplying neither raises ValueError.

Core concepts

Dynamic resources

Any attribute access on the client (that doesn't start with _) returns a Resource bound to that path segment. Chaining attributes builds up a nested REST path:

client.hosts                     # -> /api/v2/hosts
client.hosts(5)                  # -> /api/v2/hosts/5
client.hosts(5).interfaces       # -> /api/v2/hosts/5/interfaces
client.smart_proxies.list()      # -> GET /api/v2/smart_proxies

Calling a Resource with an id (client.hosts(5)) scopes it to that specific record, letting you keep chaining sub-resources under it.

CRUD methods

Every Resource exposes:

Examples:

# List all hosts (auto-paginated)
hosts = client.hosts.list()

# List with a Foreman search string
web_hosts = client.hosts.list(search="name ~ web")

# Get a single host by id
host = client.hosts.get(5)
# equivalent:
host = client.hosts(5).get()

# Create
client.hosts.create(name="myhost", organization_id=1, location_id=1)

# Update
client.hosts.update(5, name="renamed-host")

# Delete
client.hosts.delete(5)

# Non-CRUD action, e.g. power management
client.hosts.action("power", item_id=5, power_action="cycle")
# -> POST /api/v2/hosts/5/power

Katello / content endpoints

Satellite's content-related endpoints (content views, repositories, lifecycle environments, subscriptions, errata, etc.) live under /katello/api/v2 rather than /api/v2. Access them via client.katello:

content_views = client.katello.content_views.list()
repos = client.katello.repositories.list(search="name ~ EPEL")
errata = client.katello.hosts(5).errata.list()

Explicit path access

If you need a path that doesn't map cleanly onto chained attributes (e.g. numeric-looking segments, reserved Python keywords), use .api() directly:

client.api("hosts", 5, "smart_class_parameters").list()
client.api("content_views", 3, "publish", api_root="/katello/api/v2").action("publish")

Pagination

.list() paginates automatically by walking Foreman's page/per_page/total response metadata and returns a flat list of every result:

all_hosts = client.hosts.list()          # every host, across all pages

Set paginate=False to get the raw first-page response dict instead (includes total, page, per_page, results, etc.):

first_page = client.hosts.list(paginate=False, per_page=20)

Default page size is controlled by per_page on the client (default 100), and can be overridden per call:

client = ForemanClient(url, username=..., password=..., per_page=50)
client.hosts.list(per_page=200)

For manual control over iteration (e.g. to stop early), use the client's paginate() generator directly:

for host in client.paginate("/api/v2/hosts", params={"search": "name ~ web"}):
    print(host["name"])

Field projection

fields= filters the returned record(s) down to a set of top-level keys client-side, after the full response has already been received from Satellite:

client.hosts.get(5, fields=["id", "name", "operatingsystem_name"])
client.hosts.list(fields=["id", "name"])

Important: Foreman/Katello's API does not generally support arbitrary server-side field selection. This does not reduce what Satellite sends over the wire — only what's returned to your code afterward. Some endpoints support a genuine payload-reducing thin=true flag, which is a real server-side reduction and can be passed through like any other param:

client.hosts.list(thin=True)

Search syntax

Foreman/Katello search params are passed straight through as query params — use Foreman's native search syntax:

client.hosts.list(search="organization_id=1 and os_name=RedHat")
client.katello.errata.list(search="type=security and severity=critical")

Error handling

All non-2xx responses raise a typed exception from foreman_client.exceptions:

from foreman_client import (
    ForemanError,             # base class for all of the below
    ForemanConnectionError,   # network-level failure (connect/timeout)
    ForemanAuthError,         # 401 / 403
    ForemanNotFoundError,     # 404
    ForemanValidationError,   # 422 -- has an .errors attribute with Foreman's error payload
    ForemanAPIError,          # any other non-2xx
)

try:
    client.hosts.create(name="")  # missing required fields
except ForemanValidationError as exc:
    print(exc.status_code)   # 422
    print(exc.errors)        # Foreman's structured error detail
except ForemanAuthError as exc:
    print("auth problem:", exc.status_code)
except ForemanError as exc:
    print("something else went wrong:", exc)

All of these exceptions (except ForemanConnectionError) expose .status_code and .response_body.

Concurrency

The client is synchronous, but requests.Session is thread-safe for concurrent reads, so fan-out patterns work well for bulk operations across many hosts:

from concurrent.futures import ThreadPoolExecutor, as_completed

host_ids = [1, 2, 3, 4, 5]

def get_errata(host_id):
    return host_id, client.katello.hosts(host_id).errata.list()

results = {}
with ThreadPoolExecutor(max_workers=10) as pool:
    futures = {pool.submit(get_errata, hid): hid for hid in host_ids}
    for future in as_completed(futures):
        host_id, errata = future.result()
        results[host_id] = errata

SSL verification

Verification is enabled by default. Disable only if you understand the risk (e.g. internal lab Satellite with a self-signed cert):

client = ForemanClient(url, username=..., password=..., verify_ssl=False)

Project structure