REST API Client¶

Config-driven REST API client: define your APIs in YAML, call them from CLI or Python.

Overview¶

The rapi module provides a declarative approach to REST API calls. Instead of hardcoding URLs, headers, and authentication in your code, you define everything in configuration files and call endpoints by name.

from kstlib.rapi import call, RapiClient

# Quick call using config
response = call("github.user")
print(response.data)  # {"login": "octocat", ...}

# Client instance for multiple calls
client = RapiClient()
response = client.call("github.repos-issues", owner="KaminoU", repo="igcv3")

Benefits:

Single source of truth: API definitions live in YAML, not scattered in code
CLI and Python: Same config works for both kstlib rapi CLI and Python code
Credential management: Automatic token resolution from SOPS, env vars, or files
Override at runtime: Default query params can be overridden per-call

Configuration¶

In kstlib.conf.yml¶

Define APIs in your main configuration file:

# kstlib.conf.yml
rapi:
  apis:
    github:
      base_url: "https://api.github.com"
      credentials: github_token
      auth_type: bearer
      headers:
        Accept: "application/vnd.github+json"
        User-Agent: "my-app/1.0"
      endpoints:
        user:
          path: "/user"
        repos-list:
          path: "/user/repos"
          query:
            sort: updated
            per_page: "10"
        repos-issues:
          path: "/repos/{owner}/{repo}/issues"
          method: GET

External *.rapi.yml Files¶

For larger projects, define APIs in separate files:

# github.rapi.yml
name: github
base_url: "https://api.github.com"

credentials:
  type: sops
  config: github_token

headers:
  Accept: "application/vnd.github+json"
  User-Agent: "my-app/1.0"

endpoints:
  user:
    path: "/user"
  repos-list:
    path: "/user/repos"
    query:
      sort: updated
      per_page: "10"
  repos-issues:
    path: "/repos/{owner}/{repo}/issues"

Load external files via include patterns:

# kstlib.conf.yml
rapi:
  include:
    - "*.rapi.yml"
    - "apis/*.rapi.yml"

Or load directly in Python:

client = RapiClient.from_file("github.rapi.yml")
client = RapiClient.discover()  # Auto-discover *.rapi.yml in current directory

Named Server Profiles¶

The rapi.servers section is for heterogeneous APIs (different stacks, different auth) bundled into a single project. Each profile inherits from rapi.defaults via deep merge and overrides what matters (base_url, credentials, auth, headers).

rapi:
  defaults:
    auth: bearer
    headers:
      Accept: application/json

  servers:
    github:
      base_url: https://api.github.com
      credentials:
        type: env
        var: GITHUB_TOKEN
      headers:
        Accept: application/vnd.github+json
        X-GitHub-Api-Version: "2022-11-28"
    jira:
      base_url: https://mycompany.atlassian.net
      auth: api_key
      credentials:
        type: env
        var: JIRA_API_KEY

manager = load_rapi_config()
github  = manager.resolve_server("github")  # Merged: github + defaults
jira    = manager.resolve_server("jira")    # Merged: jira   + defaults
default = manager.resolve_server(None)      # Defaults wrapped as ServerConfig
manager.server_names                        # ["github", "jira"]

Note

For same API, different environments (e.g. SAS Viya source/target migration), do not use rapi.servers. Use the ${VAR} env var pattern in rapi.defaults instead - the substitution is applied globally at YAML load time, so token_path supports dynamic profile selection natively. See REST API Client for the decision matrix and full examples.

See REST API Client for full details, the decision matrix, and validation rules.

Endpoints and Parameters¶

Path Parameters¶

Use {param} placeholders in the path:

endpoints:
  repos-issues:
    path: "/repos/{owner}/{repo}/issues"

# Python: pass as keyword arguments
response = client.call("github.repos-issues", owner="KaminoU", repo="igcv3")

# CLI: pass as key=value
kstlib rapi github.repos-issues owner=KaminoU repo=igcv3

Query Parameters¶

Default Values¶

Define default query parameters in YAML:

endpoints:
  repos-list:
    path: "/user/repos"
    query:
      sort: updated
      per_page: "10"

Override at Runtime¶

Runtime arguments override YAML defaults:

# Uses defaults: sort=updated, per_page=10
response = client.call("github.repos-list")

# Override per_page, keep sort=updated
response = client.call("github.repos-list", per_page="50")

# Override both
response = client.call("github.repos-list", sort="created", per_page="100")

# CLI: same behavior
kstlib rapi github.repos-list                    # defaults
kstlib rapi github.repos-list per_page=50        # override one
kstlib rapi github.repos-list sort=created per_page=100  # override both

Pagination¶

For APIs with pagination, override the page parameter:

kstlib rapi github.repos-list page=1
kstlib rapi github.repos-list page=2
kstlib rapi github.repos-list page=3

Request Body¶

Python¶

response = client.call("myapi.create-item", body={"name": "test", "value": 42})

CLI¶

# Inline JSON
kstlib rapi myapi.create-item --body '{"name": "test", "value": 42}'

# From file (curl-like syntax)
kstlib rapi myapi.create-item --body @data.json
kstlib rapi myapi.create-item -b @data.json

CLI Usage¶

Basic Syntax¶

kstlib rapi <api>.<endpoint> [key=value ...]

Tip

Shortcut for single-API projects: The <api>. prefix is always required for clarity and script portability. If you work frequently with one API, create a shell alias:

PowerShell: function brapi { kstlib rapi binance.$args }
Bash/Zsh: alias brapi='kstlib rapi binance.'

Then use: brapi balance instead of kstlib rapi binance.balance

Examples¶

# Simple GET
kstlib rapi github.user

# With path parameters
kstlib rapi github.repos-issues owner=KaminoU repo=igcv3

# With query parameters (override defaults)
kstlib rapi github.repos-list per_page=50 page=2

# POST with body from file
kstlib rapi myapi.create-item -b @payload.json

# Custom headers
kstlib rapi github.user -H "X-Debug: true"

# Output to file (for scripting)
kstlib rapi github.user -o user.json

# Full response with metadata
kstlib rapi github.user -f full -o result.json

# List available endpoints
kstlib rapi list
kstlib rapi list github  # filter by API
kstlib rapi list --verbose  # show methods and auth

Options¶

Option	Short	Description
`--format`	`-f`	Output format: `json` (default), `text`, `full`
`--out`	`-o`	Write output to file
`--body`	`-b`	JSON body or `@filename`
`--header`	`-H`	Custom header (repeatable)
`--server`	`-s`	Named server profile from `rapi.servers` config
`--quiet`	`-q`	Suppress status messages
`--raw`		Output raw JSON without Rich formatting (pipeable)
`--minify`		Output compact single-line JSON

Server Profile Selection¶

The --server (or -s) flag picks a named profile from the rapi.servers section of kstlib.conf.yml. It overrides any server: directive declared in the *.rapi.yml file (file or endpoint level). Use it for heterogeneous APIs (github + jira + …).

# Use the github profile (overrides any server: directive in YAML)
kstlib rapi --server github github.repos-list

# Short form
kstlib rapi -s jira jira.issues-search

# Unknown server name exits 1 with the list of available profiles
kstlib rapi --server ghost github.user
# -> Error: Server profile not found: 'ghost'. Available: ['github', 'jira']

For SAS Viya source/target migration (same API, different environments), use the ${ACTIVE_VIYA} env var pattern instead of --server. See REST API Client for the decision matrix.

Verbosity and Tracing¶

kstlib rapi github.user          # normal output
kstlib -v rapi github.user       # debug logging
kstlib -vv rapi github.user      # verbose debug
kstlib -vvv rapi github.user     # TRACE: shows body, params, elapsed time

Python API¶

Quick Functions¶

from kstlib.rapi import call, call_async

# Synchronous
response = call("github.user")

# Asynchronous
response = await call_async("github.user")

Client Instance¶

from kstlib.rapi import RapiClient

# From kstlib.conf.yml
client = RapiClient()

# From external file
client = RapiClient.from_file("github.rapi.yml")

# Auto-discover *.rapi.yml files
client = RapiClient.discover()

# Make calls
response = client.call("github.user")
response = client.call("github.repos-issues", owner="KaminoU", repo="igcv3")

# With body
response = client.call("myapi.create", body={"key": "value"})

# With custom headers
response = client.call("myapi.get", headers={"X-Custom": "value"})

Response Object¶

response = client.call("github.user")

response.ok          # True if 2xx status
response.status_code # HTTP status code
response.data        # Parsed JSON (dict or list)
response.text        # Raw response text
response.headers     # Response headers
response.elapsed     # Request duration in seconds
response.endpoint_ref # "github.user"

Credentials¶

Configuration¶

# In kstlib.conf.yml
credentials:
  github_token:
    type: sops
    path: "secrets/github.sops.json"
    key: "access_token"

  api_key:
    type: env
    key: "MY_API_KEY"

rapi:
  apis:
    github:
      credentials: github_token  # Reference by name

Inline Credentials (*.rapi.yml)¶

# github.rapi.yml
credentials:
  type: sops
  config: github_token  # References credentials section in kstlib.conf.yml

Supported Sources¶

Type	Parameters	Description
`sops`	`path`, `fields`	SOPS-encrypted files
`env`	`fields`	Environment variables
`file`	`path`, `fields`	Plain text/JSON files
`keyring`	`service`, `username`	System keyring

Generic Fields Mapping¶

All credential types support a generic fields: mapping for flexible credential extraction. This allows APIs requiring more than key/secret (e.g., passphrase for Coinbase/KuCoin/OKX).

# Generic fields mapping (recommended)
credentials:
  type: sops  # or env, file
  path: "./tokens/exchange.sops.yml"
  fields:
    key: api_key           # Required - maps to CredentialRecord.value
    secret: api_secret     # Optional - maps to CredentialRecord.secret
    passphrase: passphrase # Optional - maps to CredentialRecord.extras["passphrase"]
    account_id: account    # Optional - any extra field goes to extras dict

Field mapping rules:

key is required and maps to CredentialRecord.value
secret is optional and maps to CredentialRecord.secret
All other fields map to CredentialRecord.extras dict

Environment variables example:

credentials:
  type: env
  fields:
    key: COINBASE_API_KEY
    secret: COINBASE_API_SECRET
    passphrase: COINBASE_PASSPHRASE

Legacy format (still supported):

# Legacy - key_field/secret_field (backwards compatible)
credentials:
  type: sops
  path: "./tokens/binance.sops.yml"
  key_field: api_key
  secret_field: secret_key

HMAC Authentication¶

For APIs requiring HMAC signature (Binance, Kraken, etc.), configure the auth section:

# binance.rapi.yml
name: binance
base_url: "https://testnet.binance.vision"

credentials:
  type: sops
  path: "./tokens/binance.sops.yml"
  fields:
    key: api_key
    secret: secret_key

auth:
  type: hmac
  algorithm: sha256          # sha256 (Binance) or sha512 (Kraken)
  timestamp_field: timestamp # Query param name for timestamp
  signature_field: signature # Query param name for signature
  signature_format: hex      # hex (Binance) or base64 (Kraken)
  key_header: X-MBX-APIKEY   # Header for API key

endpoints:
  balance:
    path: "/api/v3/account"
    method: GET

  my-trades:
    path: "/api/v3/myTrades"
    method: GET
    query:
      symbol: BTCUSDT
      limit: "10"

HMAC Options¶

Option	Description	Example
`algorithm`	Hash algorithm	`sha256`, `sha512`
`timestamp_field`	Query param for timestamp (ms)	`timestamp`, `nonce`
`signature_field`	Query param for signature	`signature`, `sign`
`signature_format`	Signature encoding	`hex`, `base64`
`key_header`	Header for API key	`X-MBX-APIKEY`, `API-Key`
`sign_body`	Sign request body instead of query string	`true`, `false`
`nonce_field`	Alternative to timestamp_field	`nonce`

Public Endpoints (No Auth)¶

For APIs with mixed public/private endpoints, disable auth per-endpoint with auth: false:

endpoints:
  # Public - no signature needed
  ticker-price:
    path: "/api/v3/ticker/price"
    method: GET
    auth: false
    query:
      symbol: BTCUSDT

  # Private - HMAC signature applied
  balance:
    path: "/api/v3/account"
    method: GET
    # auth: true (default)

Note

By default, all endpoints inherit API-level authentication (auth: true). Set auth: false on public endpoints to skip credential resolution and signature generation.

Safeguards for Dangerous Endpoints¶

For destructive operations (DELETE, PUT), kstlib requires explicit confirmation to prevent accidental data loss.

Configuration¶

Define a safeguard string on dangerous endpoints:

endpoints:
  delete-user:
    path: "/users/{user_id}"
    method: DELETE
    safeguard: "DELETE USER {user_id}"

By default, DELETE and PUT methods require a safeguard. Configure this in kstlib.conf.yml:

rapi:
  safeguard:
    required_methods:
      - DELETE
      - PUT

Calling Safeguarded Endpoints¶

from kstlib.rapi import RapiClient
from kstlib.rapi.exceptions import ConfirmationRequiredError

client = RapiClient()

# Without confirmation - raises ConfirmationRequiredError
try:
    client.call("admin.delete-user", user_id="123")
except ConfirmationRequiredError as e:
    print(f"Required: {e.expected}")  # "DELETE USER 123"

# With correct confirmation - proceeds
client.call("admin.delete-user", user_id="123", confirm="DELETE USER 123")

Warning

If an endpoint uses a method in required_methods but lacks a safeguard string, SafeguardMissingError is raised at config load time (not at runtime).

Exchange Examples¶

Binance (SHA256, hex, query string):

auth:
  type: hmac
  algorithm: sha256
  timestamp_field: timestamp
  signature_field: signature
  signature_format: hex
  key_header: X-MBX-APIKEY

Kraken (SHA512, base64, nonce):

auth:
  type: hmac
  algorithm: sha512
  nonce_field: nonce
  signature_field: sign
  signature_format: base64
  key_header: API-Key

# Test HMAC signing
cd examples/rapi/binance
kstlib rapi binance.balance        # Signed request
kstlib -vvv rapi binance.balance   # TRACE mode to see signature details

API Reference¶

Client¶

class kstlib.rapi.RapiClient(config_manager=None, credentials_config=None, *, ssl_verify=None, ssl_ca_bundle=None)[source]¶

Bases: object

Config-driven REST API client.

Makes HTTP requests to configured API endpoints with automatic credential resolution, header merging, and detailed logging.

Supports loading configuration from: - kstlib.conf.yml (default) - External *.rapi.yml files (via from_file) - Auto-discovery of *.rapi.yml in current directory (via discover)

Parameters:

config_manager (RapiConfigManager | None) – Optional RapiConfigManager (loads from config if None).
credentials_config (Mapping[str, Any] | None) – Optional credentials configuration.

Examples

>>> client = RapiClient()  
>>> response = client.call("httpbin.get_ip")  
>>> response.data  
{'origin': '...'}

>>> client = RapiClient.from_file("github.rapi.yml")  
>>> client = RapiClient.discover()  

__init__(self, config_manager: 'RapiConfigManager | None' = None, credentials_config: 'Mapping[str, Any] | None' = None, *, ssl_verify: 'bool | None' = None, ssl_ca_bundle: 'str | None' = None) → 'None' -> None[source]¶

Initialize RapiClient.

Parameters:

config_manager (RapiConfigManager | None) – Optional RapiConfigManager instance.
credentials_config (Mapping[str, Any] | None) – Optional credentials configuration.
ssl_verify (bool | None) – Override SSL verification (True/False). If None, uses global config from kstlib.conf.yml.
ssl_ca_bundle (str | None) – Override CA bundle path. If None, uses global config from kstlib.conf.yml.

classmethod from_file(path: 'str', credentials_config: 'Mapping[str, Any] | None' = None) → 'RapiClient' -> RapiClient[source]¶

Create client from a *.rapi.yml file.

Loads API configuration from an external YAML file with simplified format.

Parameters:

path (str) – Path to the *.rapi.yml file.
credentials_config (Mapping[str, Any] | None) – Additional credentials (merged with inline).

Returns:

Configured RapiClient instance.

Raises:

FileNotFoundError – If file does not exist.
ValueError – If file format is invalid.

Return type:

RapiClient

Examples

>>> client = RapiClient.from_file("github.rapi.yml")  
>>> response = client.call("github.user")  

classmethod discover(directory: 'str | None' = None, pattern: 'str' = '*.rapi.yml', credentials_config: 'Mapping[str, Any] | None' = None) → 'RapiClient' -> RapiClient[source]¶

Create client by auto-discovering *.rapi.yml files.

Searches for files matching the pattern in the specified directory (defaults to current working directory) and loads all found configs.

Parameters:

directory (str | None) – Directory to search in (default: current directory).
pattern (str) – Glob pattern for files (default: *.rapi.yml).
credentials_config (Mapping[str, Any] | None) – Additional credentials (merged with inline).

Returns:

Configured RapiClient instance.

Raises:

FileNotFoundError – If no matching files found.

Return type:

RapiClient

Examples

>>> client = RapiClient.discover()  
>>> client = RapiClient.discover("./apis/")  

property config_manager: RapiConfigManager¶

Get the configuration manager.

Returns:: RapiConfigManager instance.

list_apis(self) → 'list[str]' -> list[str][source]¶

List all configured API names.

Returns:: List of API names.
Return type:: list[str]

list_endpoints(self, api_name: 'str | None' = None) → 'list[str]' -> list[str][source]¶

List endpoint references.

Parameters:: api_name (str | None) – Filter by API name (optional).
Returns:: List of full endpoint references (api.endpoint).
Return type:: list[str]

call(self, endpoint_ref: 'str', *args: 'Any', body: 'Any' = None, headers: 'Mapping[str, str] | None' = None, timeout: 'float | None' = None, confirm: 'str | None' = None, server: 'str | None' = None, **kwargs: 'Any') → 'RapiResponse' -> RapiResponse[source]¶

Make a synchronous API call.

Parameters:

endpoint_ref (str) – Endpoint reference (full: api.endpoint or short: endpoint).
*args (Any) – Positional arguments for path parameters.
body (Any) – Request body (dict for JSON, str for raw).
headers (Mapping[str, str] | None) – Runtime headers (override service/endpoint headers).
timeout (float | None) – Request timeout (uses config default if None).
confirm (str | None) – Confirmation string for dangerous endpoints with safeguard.
server (str | None) – Optional named server profile from rapi.servers to use for this call. Overrides any server: directive set at the endpoint or file level in the YAML config. Cascade: runtime server > endpoint server: > file server: > static ApiConfig (no server).
**kwargs (Any) – Keyword arguments for path parameters and query params.

Returns:

RapiResponse with parsed data.

Raises:

AuthExpiredError – If the response signals access token expiration (HTTP 401 with body keyword or WWW-Authenticate invalid_token marker). Terminal, not retried.
ConfirmationRequiredError – If safeguard requires confirmation.
RequestError – If request fails after retries.
ResponseTooLargeError – If response exceeds max size.
ServerNotFoundError – If server (or any cascading directive) does not exist in rapi.servers.

Return type:

RapiResponse

Examples

>>> client = RapiClient()  
>>> client.call("httpbin.get_ip")  
>>> client.call("httpbin.delayed", 5)  
>>> client.call("httpbin.post_data", body={"key": "value"})  
>>> client.call("admin.delete_user", userId="123", confirm="DELETE USER 123")  
>>> client.call("github.repos-list", server="github")  

async call_async(self, endpoint_ref: 'str', *args: 'Any', body: 'Any' = None, headers: 'Mapping[str, str] | None' = None, timeout: 'float | None' = None, confirm: 'str | None' = None, server: 'str | None' = None, **kwargs: 'Any') → 'RapiResponse' -> RapiResponse[source]¶

Make an asynchronous API call.

Parameters:

endpoint_ref (str) – Endpoint reference (full: api.endpoint or short: endpoint).
*args (Any) – Positional arguments for path parameters.
body (Any) – Request body (dict for JSON, str for raw).
headers (Mapping[str, str] | None) – Runtime headers (override service/endpoint headers).
timeout (float | None) – Request timeout (uses config default if None).
confirm (str | None) – Confirmation string for dangerous endpoints with safeguard.
server (str | None) – Optional named server profile from rapi.servers to use for this call. See call() for cascade rules.
**kwargs (Any) – Keyword arguments for path parameters and query params.

Returns:

RapiResponse with parsed data.

Raises:

AuthExpiredError – If the response signals access token expiration (HTTP 401 with body keyword or WWW-Authenticate invalid_token marker). Terminal, not retried.
ConfirmationRequiredError – If safeguard requires confirmation.
RequestError – If request fails after retries.
ResponseTooLargeError – If response exceeds max size.
ServerNotFoundError – If server (or any cascading directive) does not exist in rapi.servers.

Return type:

RapiResponse

class kstlib.rapi.RapiResponse(status_code, headers=<factory>, data=None, text='', elapsed=0.0, endpoint_ref='')[source]

Bases: object

Response from an API call.

status_code

HTTP status code.

Type:: int

headers

Response headers.

Type:: dict[str, str]

data

Parsed JSON response (or None if not JSON).

Type:: Any

text

Raw response text.

Type:: str

elapsed

Request duration in seconds.

Type:: float

endpoint_ref

Full endpoint reference used.

Type:: str

Examples

>>> response = RapiResponse(status_code=200, data={"ip": "1.2.3.4"})
>>> response.ok
True
>>> response.data["ip"]
'1.2.3.4'

status_code: int

headers: dict[str, str]

data: Any = None

text: str = ''

elapsed: float = 0.0

endpoint_ref: str = ''

property ok: bool: Return True if status code indicates success (2xx).

__init__(self, status_code: 'int', headers: 'dict[str, str]' = <factory>, data: 'Any' = None, text: 'str' = '', elapsed: 'float' = 0.0, endpoint_ref: 'str' = '') → None -> None

class kstlib.rapi.FilePayload(filename, data, content_type, field_name='file')[source]

Bases: object

Carrier for file upload data (multipart mode).

Use this to upload file content programmatically without a file on disk.

filename

Original filename (used in Content-Disposition).

Type:: str

data

Raw file bytes.

Type:: bytes

content_type

MIME type of the file.

Type:: str

field_name

Form field name for the file part.

Type:: str

Examples

>>> payload = FilePayload(
...     filename="report.csv",
...     data=b"col1,col2",
...     content_type="text/csv",
... )
>>> payload.field_name
'file'

filename: str

data: bytes

content_type: str

field_name: str = 'file'

__init__(self, filename: 'str', data: 'bytes', content_type: 'str', field_name: 'str' = 'file') → None -> None

kstlib.rapi.call(endpoint_ref: 'str', *args: 'Any', body: 'Any' = None, headers: 'Mapping[str, str] | None' = None, confirm: 'str | None' = None, server: 'str | None' = None, **kwargs: 'Any') → 'RapiResponse' -> RapiResponse[source]¶

Make a quick synchronous API call using a temporary RapiClient.

Creates a temporary RapiClient and makes the call.

Parameters:

endpoint_ref (str) – Endpoint reference.
*args (Any) – Positional path parameters.
body (Any) – Request body.
headers (Mapping[str, str] | None) – Runtime headers.
confirm (str | None) – Confirmation string for dangerous endpoints with safeguard.
server (str | None) – Optional named server profile from rapi.servers (see RapiClient.call() for cascade rules).
**kwargs (Any) – Keyword parameters.

Returns:

RapiResponse.

Return type:

RapiResponse

Examples

>>> from kstlib.rapi import call  
>>> response = call("httpbin.get_ip")  
>>> response = call("github.repos-list", server="github")  

async kstlib.rapi.call_async(endpoint_ref: 'str', *args: 'Any', body: 'Any' = None, headers: 'Mapping[str, str] | None' = None, confirm: 'str | None' = None, server: 'str | None' = None, **kwargs: 'Any') → 'RapiResponse' -> RapiResponse[source]¶

Make a quick asynchronous API call using a temporary RapiClient.

Creates a temporary RapiClient and makes the async call.

Parameters:

endpoint_ref (str) – Endpoint reference.
*args (Any) – Positional path parameters.
body (Any) – Request body.
headers (Mapping[str, str] | None) – Runtime headers.
confirm (str | None) – Confirmation string for dangerous endpoints with safeguard.
server (str | None) – Optional named server profile from rapi.servers (see RapiClient.call() for cascade rules).
**kwargs (Any) – Keyword parameters.

Returns:

RapiResponse.

Return type:

RapiResponse

Examples

>>> from kstlib.rapi import call_async  
>>> response = await call_async("httpbin.get_ip")  

Configuration¶

class kstlib.rapi.RapiConfigManager(rapi_config=None, credentials_config=None, safeguard_config=None, strict=False)[source]¶

Bases: object

Manage RAPI configuration and endpoint resolution.

Loads API and endpoint configurations from kstlib.conf.yml and provides resolution methods supporting both full references (api.endpoint) and short references (endpoint only, auto-resolved if unique).

Supports loading from: - kstlib.conf.yml (default) - External *.rapi.yml files (via from_file/from_files) - Auto-discovery of *.rapi.yml in current directory (via discover)

Parameters:

rapi_config (Mapping[str, Any] | None) – The ‘rapi’ section from configuration.
credentials_config (Mapping[str, Any] | None) – Inline credentials extracted from *.rapi.yml files.

Examples

>>> manager = RapiConfigManager({"api": {"httpbin": {"base_url": "..."}}})
>>> endpoint = manager.resolve("httpbin.get_ip")  

>>> manager = RapiConfigManager.from_file("github.rapi.yml")  
>>> manager = RapiConfigManager.discover()  

__init__(self, rapi_config: 'Mapping[str, Any] | None' = None, credentials_config: 'Mapping[str, Any] | None' = None, safeguard_config: 'SafeguardConfig | None' = None, strict: 'bool' = False) → 'None' -> None[source]¶

Initialize RapiConfigManager.

Parameters:

rapi_config (Mapping[str, Any] | None) – The ‘rapi’ section from configuration.
credentials_config (Mapping[str, Any] | None) – Inline credentials from *.rapi.yml files.
safeguard_config (SafeguardConfig | None) – Safeguard configuration (default: DELETE and PUT require safeguard).
strict (bool) – If True, raise error on endpoint collisions. If False, warn and overwrite.

classmethod from_file(path: 'str | Path', base_dir: 'Path | None' = None, safeguard_config: 'SafeguardConfig | None' = None, defaults: 'dict[str, Any] | None' = None, strict: 'bool' = False) → 'RapiConfigManager' -> RapiConfigManager[source]¶

Load configuration from a single *.rapi.yml file.

The file format is simplified compared to kstlib.conf.yml, with top-level keys: name, base_url, credentials, auth, headers, endpoints.

Parameters:

path (str | Path) – Path to the *.rapi.yml file.
base_dir (Path | None) – Base directory for resolving relative paths in credentials.
safeguard_config (SafeguardConfig | None) – Safeguard configuration (default: DELETE and PUT require safeguard).
defaults (dict[str, Any] | None) – Default values inherited from kstlib.conf.yml rapi.defaults section.
strict (bool) – If True, raise error on endpoint collisions. If False, warn and overwrite.

Returns:

Configured RapiConfigManager instance.

Raises:

FileNotFoundError – If file does not exist.
ValueError – If file format is invalid.

Return type:

RapiConfigManager

Examples

>>> manager = RapiConfigManager.from_file("github.rapi.yml")  

classmethod from_files(paths: 'Sequence[str | Path]', base_dir: 'Path | None' = None, safeguard_config: 'SafeguardConfig | None' = None, defaults: 'dict[str, Any] | None' = None, strict: 'bool' = False) → 'RapiConfigManager' -> RapiConfigManager[source]¶

Load configuration from multiple *.rapi.yml files.

Parameters:

paths (Sequence[str | Path]) – List of paths to *.rapi.yml files.
base_dir (Path | None) – Base directory for resolving relative paths.
safeguard_config (SafeguardConfig | None) – Safeguard configuration (default: DELETE and PUT require safeguard).
defaults (dict[str, Any] | None) – Default values inherited from kstlib.conf.yml rapi.defaults section. Supports: base_url, credentials, auth, headers.
strict (bool) – If True, raise error on endpoint collisions. If False, warn and overwrite.

Returns:

Configured RapiConfigManager instance with merged configs.

Raises:

FileNotFoundError – If any file does not exist.
ValueError – If any file format is invalid.
EndpointCollisionError – If strict=True and endpoints collide.

Return type:

RapiConfigManager

Examples

>>> manager = RapiConfigManager.from_files([
...     "github.rapi.yml",
...     "slack.rapi.yml",
... ])  

classmethod discover(directory: 'str | Path | None' = None, pattern: 'str' = '*.rapi.yml') → 'RapiConfigManager' -> RapiConfigManager[source]¶

Auto-discover and load *.rapi.yml files from a directory.

Searches for files matching the pattern in the specified directory (defaults to current working directory).

Parameters:

directory (Path | str | None) – Directory to search in (default: current directory).
pattern (str) – Glob pattern for files (default: *.rapi.yml).

Returns:

Configured RapiConfigManager instance.

Raises:

FileNotFoundError – If no matching files found.

Return type:

RapiConfigManager

Examples

>>> manager = RapiConfigManager.discover()  
>>> manager = RapiConfigManager.discover("./apis/")  

property credentials_config: dict[str, Any]¶

Get inline credentials config extracted from *.rapi.yml files.

Returns:: Dictionary of credentials configurations.

property source_files: list[Path]¶

Get list of source files loaded.

Returns:: List of Path objects for loaded files.

property safeguard_config: SafeguardConfig¶

Get safeguard configuration.

Returns:: SafeguardConfig instance.

resolve_server(self, server_name: 'str | None' = None) → 'ServerConfig' -> ServerConfig[source]¶

Resolve a named server profile, merged with defaults.

If server_name is None, returns the defaults as a ServerConfig. If server_name is given, merges rapi.servers.<name> on top of rapi.defaults using deep merge (server values win).

Parameters:: server_name (str | None) – Named server profile, or None for defaults.
Returns:: Resolved ServerConfig with merged values.
Raises:: ServerNotFoundError – If server_name is not in rapi.servers.
Return type:: ServerConfig

Examples

>>> manager = load_rapi_config()  
>>> server = manager.resolve_server("source")  
>>> server.base_url  
'https://viya-source.example.com'

property server_names: list[str]¶

Get list of configured server profile names.

Returns:: List of server names from rapi.servers section.

resolve_effective_server(self, api_config: 'ApiConfig', endpoint_config: 'EndpointConfig', runtime_server: 'str | None' = None) → 'ServerConfig | None' -> ServerConfig | None[source]¶

Resolve the effective server profile for a given request.

Cascade priority (highest to lowest):

runtime_server (e.g. CLI --server flag or call(server=...))
endpoint_config.server (endpoint-level server: directive)
api_config.server (file-level server: directive)
None (caller falls back to the static api_config)

Parameters:

api_config (ApiConfig) – API configuration for the called endpoint.
endpoint_config (EndpointConfig) – Endpoint configuration for the called endpoint.
runtime_server (str | None) – Optional runtime override (CLI flag or kwarg).

Returns:

Resolved ServerConfig if any cascade level provided a name, otherwise None (caller should use the static ApiConfig).

Raises:

ServerNotFoundError – If the resolved name does not exist in rapi.servers.

Return type:

ServerConfig | None

resolve(self, endpoint_ref: 'str') → 'tuple[ApiConfig, EndpointConfig]' -> tuple[ApiConfig, EndpointConfig][source]¶

Resolve endpoint reference to configuration.

Supports both full references (api.endpoint) and short references (endpoint only). Short references are auto-resolved if the endpoint name is unique across all APIs.

Parameters:

endpoint_ref (str) – Full reference (api.endpoint) or short (endpoint).

Returns:

Tuple of (ApiConfig, EndpointConfig).

Raises:

EndpointNotFoundError – If endpoint cannot be found.
EndpointAmbiguousError – If short reference matches multiple APIs.

Return type:

tuple[ApiConfig, EndpointConfig]

Examples

>>> manager = RapiConfigManager({...})  
>>> api, endpoint = manager.resolve("httpbin.get_ip")  
>>> api, endpoint = manager.resolve("get_ip")  

get_api(self, api_name: 'str') → 'ApiConfig | None' -> ApiConfig | None[source]¶

Get API configuration by name.

Parameters:: api_name (str) – API service name.
Returns:: ApiConfig or None if not found.
Return type:: ApiConfig | None

list_apis(self) → 'list[str]' -> list[str][source]¶

List all configured API names.

Returns:: List of API names.
Return type:: list[str]

property apis: dict[str, ApiConfig]¶

Get all configured APIs.

Returns:: Dictionary mapping API names to ApiConfig objects.

list_endpoints(self, api_name: 'str | None' = None) → 'list[str]' -> list[str][source]¶

List endpoint references.

Parameters:: api_name (str | None) – Filter by API name (optional).
Returns:: List of full endpoint references.
Return type:: list[str]

class kstlib.rapi.ApiConfig(name, base_url, credentials=None, auth_type=None, hmac_config=None, headers=<factory>, endpoints=<factory>, server=None)[source]

Bases: object

Configuration for an API service.

name

API service name (e.g., “httpbin”).

Type:: str

base_url

Base URL for the API.

Type:: str

credentials

Name of credential config to use.

Type:: str | None

auth_type

Authentication type (bearer, basic, api_key, hmac).

Type:: str | None

hmac_config

HMAC signing configuration (required when auth_type is hmac).

Type:: kstlib.rapi.config.HmacConfig | None

headers

Service-level headers (applied to all endpoints).

Type:: dict[str, str]

endpoints

Dictionary of endpoint configurations.

Type:: dict[str, kstlib.rapi.config.EndpointConfig]

server

Optional named server profile this API file should use by default. Resolved against rapi.servers at request time. Acts as a fallback when individual endpoints do not declare their own server: directive. Validated at config-load time when rapi.servers is present.

Type:: str | None

Examples

>>> api = ApiConfig(
...     name="httpbin",
...     base_url="https://httpbin.org",
...     endpoints={},
... )

name: str

base_url: str

credentials: str | None

auth_type: str | None

hmac_config: HmacConfig | None

headers: dict[str, str]

endpoints: dict[str, EndpointConfig]

server: str | None

__init__(self, name: 'str', base_url: 'str', credentials: 'str | None' = None, auth_type: 'str | None' = None, hmac_config: 'HmacConfig | None' = None, headers: 'dict[str, str]' = <factory>, endpoints: 'dict[str, EndpointConfig]' = <factory>, server: 'str | None' = None) → None -> None

class kstlib.rapi.EndpointConfig(name, api_name, path, method='GET', query=<factory>, headers=<factory>, body_template=None, auth=True, safeguard=None, description=None, multipart=None, server=None)[source]

Bases: object

Configuration for a single API endpoint.

name

Endpoint name (e.g., “get_ip”).

Type:: str

api_name

Parent API name (e.g., “httpbin”).

Type:: str

path

URL path template (e.g., “/delay/{seconds}”).

Type:: str

method

HTTP method (GET, POST, PUT, DELETE, PATCH).

Type:: str

query

Default query parameters.

Type:: dict[str, str | None]

headers

Endpoint-level headers (merged with service headers).

Type:: dict[str, str]

body_template

Default body template for POST/PUT.

Type:: dict[str, Any] | None

auth

Whether to apply API-level authentication to this endpoint. Set to False for public endpoints that don’t require auth.

Type:: bool

description

Human-readable description of the endpoint.

Type:: str | None

server

Optional named server profile this endpoint should use. Resolved against rapi.servers at request time. Overrides any server: directive set at the file level on the parent ApiConfig. Validated at config-load time when rapi.servers is present.

Type:: str | None

Examples

>>> config = EndpointConfig(
...     name="get_ip",
...     api_name="httpbin",
...     path="/ip",
...     method="GET",
... )
>>> config.full_ref
'httpbin.get_ip'

name: str

api_name: str

path: str

method: str

query: dict[str, str | None]

headers: dict[str, str]

body_template: dict[str, Any] | None

auth: bool

safeguard: str | None

description: str | None

multipart: MultipartConfig | None

server: str | None

property is_multipart: bool: Check if endpoint is configured for multipart/form-data upload.

__post_init__(self) → 'None' -> None[source]

Validate safeguard field (deep defense).

property full_ref: str

api_name.endpoint_name.

Type:: Return full reference

build_path(self, *args: 'Any', **kwargs: 'Any') → 'str' -> str[source]

Build path with positional and keyword arguments.

Parameters:

*args (Any) – Positional arguments for {0}, {1}, etc.
**kwargs (Any) – Keyword arguments for {name} placeholders.

Returns:

Formatted path string.

Raises:

ValueError – If required parameters are missing.

Return type:

str

Examples

>>> config = EndpointConfig(
...     name="delay",
...     api_name="httpbin",
...     path="/delay/{seconds}",
... )
>>> config.build_path(seconds=5)
'/delay/5'
>>> config.build_path(5)
'/delay/5'

build_safeguard(self, *args: 'Any', **kwargs: 'Any') → 'str | None' -> str | None[source]

Build safeguard string with variable substitution.

Substitutes {param} placeholders in the safeguard string with provided arguments, similar to build_path.

Parameters:

*args (Any) – Positional arguments for {0}, {1}, etc.
**kwargs (Any) – Keyword arguments for {name} placeholders.

Returns:

Substituted safeguard string, or None if no safeguard configured.

Return type:

str | None

Examples

>>> config = EndpointConfig(
...     name="delete",
...     api_name="test",
...     path="/users/{userId}",
...     method="DELETE",
...     safeguard="DELETE USER {userId}",
... )
>>> config.build_safeguard(userId="abc123")
'DELETE USER abc123'

__init__(self, name: 'str', api_name: 'str', path: 'str', method: 'str' = 'GET', query: 'dict[str, str | None]' = <factory>, headers: 'dict[str, str]' = <factory>, body_template: 'dict[str, Any] | None' = None, auth: 'bool' = True, safeguard: 'str | None' = None, description: 'str | None' = None, multipart: 'MultipartConfig | None' = None, server: 'str | None' = None) → None -> None

class kstlib.rapi.MultipartConfig(field_name='file', content_type=None)[source]

Bases: object

Multipart upload configuration for an endpoint.

field_name

Form field name for the file (default: “file”).

Type:: str

content_type

Override MIME type (auto-detected from filename if None).

Type:: str | None

Examples

>>> config = MultipartConfig(field_name="dataFile")
>>> config.field_name
'dataFile'

field_name: str

content_type: str | None

__post_init__(self) → 'None' -> None[source]

Validate multipart config values.

__init__(self, field_name: 'str' = 'file', content_type: 'str | None' = None) → None -> None

class kstlib.rapi.ServerConfig(name, base_url, credentials=<factory>, auth=None, headers=<factory>)[source]

Bases: object

Resolved server profile (defaults merged with server overrides).

Created by RapiConfigManager.resolve_server() after merging rapi.defaults with a named rapi.servers.<name> profile.

name

Server profile name (or "defaults" for the fallback).

Type:: str

base_url

Base URL for the server.

Type:: str

credentials

Credentials configuration dict.

Type:: dict[str, Any]

auth

Authentication type string (bearer, basic, api_key, hmac).

Type:: str | None

headers

Merged headers dict.

Type:: dict[str, str]

Examples

>>> cfg = ServerConfig(
...     name="source",
...     base_url="https://viya-source.example.com",
...     credentials={"type": "file", "path": "~/.sas/creds.json"},
...     auth="bearer",
...     headers={"Accept": "application/json"},
... )

name: str

base_url: str

credentials: dict[str, Any]

auth: str | None

headers: dict[str, str]

__init__(self, name: 'str', base_url: 'str', credentials: 'dict[str, Any]' = <factory>, auth: 'str | None' = None, headers: 'dict[str, str]' = <factory>) → None -> None

kstlib.rapi.load_rapi_config() → 'RapiConfigManager' -> RapiConfigManager[source]¶

Load RAPI configuration from kstlib.conf.yml with include support.

Supports including external *.rapi.yml files via glob patterns, and a defaults section that is inherited by included files:

rapi:
  # Strict mode: error on endpoint collisions (default: false = warn only)
  strict: true

  # Defaults inherited by all included *.rapi.yml files
  defaults:
    base_url: "https://${VIYA_HOST}"
    credentials:
      type: file
      path: ~/.sas/credentials.json
      token_path: ".Default['access-token']"
    auth: bearer
    headers:
      Accept: application/json

  include:
    - "./apis/*.rapi.yml"
    - "~/.config/kstlib/*.rapi.yml"

  safeguard:
    required_methods:
      - DELETE

  api:
    httpbin:
      base_url: "https://httpbin.org"
      # ...

With defaults, included files can be minimal:

# annotations.rapi.yml
name: annotations
headers:
  Accept: application/vnd.sas.annotation+json
endpoints:
  root:
    path: /annotations/
    method: GET

Returns:: Configured RapiConfigManager instance with merged configs.
Return type:: RapiConfigManager

Examples

>>> manager = load_rapi_config()  

Credentials¶

class kstlib.rapi.CredentialResolver(credentials_config=None)[source]¶

Bases: object

Resolve credentials from multiple sources.

Supported credential types: - env: Environment variable - file: JSON/YAML file with jq-like path extraction - sops: SOPS-encrypted file - provider: kstlib.auth provider (OAuth2/OIDC)

Parameters:: credentials_config (Mapping[str, Any] | None) – Credentials section from config.

Examples

>>> resolver = CredentialResolver({"github": {"type": "env", "var": "GITHUB_TOKEN"}})
>>> record = resolver.resolve("github")  

__init__(self, credentials_config: 'Mapping[str, Any] | None' = None) → 'None' -> None[source]¶

Initialize CredentialResolver.

Parameters:: credentials_config (Mapping[str, Any] | None) – Credentials section from config.

resolve(self, credential_name: 'str') → 'CredentialRecord' -> CredentialRecord[source]¶

Resolve a credential by name.

Parameters:: credential_name (str) – Name of the credential in config.
Returns:: CredentialRecord with resolved value(s).
Raises:: CredentialError – If credential cannot be resolved.
Return type:: CredentialRecord

resolve_inline(self, cred_config: 'Mapping[str, Any]', *, name_hint: 'str' = '<inline>') → 'CredentialRecord' -> CredentialRecord[source]¶

Resolve a credential from an inline config dict (no name lookup).

Used by ServerConfig profiles where credentials are declared inline (as a dict) rather than referenced by name. Bypasses both the config registry lookup and the resolver cache, since inline configs have no stable identifier to key on.

Parameters:

cred_config (Mapping[str, Any]) – Inline credential configuration dict (with at least a type key: env, file, sops, or provider).
name_hint (str) – Label used for logging and error messages (e.g. "server.github").

Returns:

CredentialRecord with resolved value(s).

Raises:

CredentialError – If the inline config is invalid or resolution fails.

Return type:

CredentialRecord

Examples

>>> resolver = CredentialResolver()
>>> # cred = resolver.resolve_inline(
... #     {"type": "env", "var": "GITHUB_TOKEN"},
... #     name_hint="server.github",
... # )  

static extract_value(data: 'Any', path: 'str') → 'Any' -> Any[source]¶

Extract value using jq-like path syntax.

Supports: - .foo.bar - nested object access - .foo[0] - array index access - .foo[“key-with-dash”] - bracket notation for special keys - .foo[‘key-with-dash’] - single quotes also supported - .foo.bar[0].baz - combined access

Parameters:

data (Any) – Data structure to extract from.
path (str) – jq-like path (e.g., “.foo.bar[0]” or ‘.foo[“access-token”]’).

Returns:

Extracted value or None if not found.

Return type:

Any

Examples

>>> CredentialResolver.extract_value({"foo": {"bar": [1, 2, 3]}}, ".foo.bar[1]")
2
>>> CredentialResolver.extract_value({"a": "b"}, ".missing")
>>> CredentialResolver.extract_value([1, 2, 3], ".[0]")
1
>>> CredentialResolver.extract_value({"a-b": "value"}, '.["a-b"]')
'value'

clear_cache(self) → 'None' -> None[source]¶

Clear the credential cache.

class kstlib.rapi.CredentialRecord(value, secret=None, source='unknown', expires_at=None, extras=<factory>)[source]

Bases: object

Resolved credential with metadata.

value

Primary credential value (token, API key).

Type:: str

secret

Secondary credential value (API secret for signing).

Type:: str | None

source

Source type that provided this credential.

Type:: str

expires_at

Expiration timestamp (if known).

Type:: float | None

extras

Additional credential fields (passphrase, etc.).

Type:: dict[str, str]

Examples

>>> record = CredentialRecord(value="token123", source="env")
>>> record.value
'token123'
>>> record = CredentialRecord(
...     value="key", secret="secret", source="sops",
...     extras={"passphrase": "pass123"}
... )
>>> record.extras.get("passphrase")
'pass123'

value: str

secret: str | None

source: str

expires_at: float | None

extras: dict[str, str]

__init__(self, value: 'str', secret: 'str | None' = None, source: 'str' = 'unknown', expires_at: 'float | None' = None, extras: 'dict[str, str]' = <factory>) → None -> None

Exceptions¶

exception kstlib.rapi.RapiError(message, *, details=None)[source]¶

Bases: KstlibError

Base exception for all RAPI errors.

message¶: Human-readable error message.

details¶: Additional error context as key-value pairs.

Examples

>>> raise RapiError("Something went wrong", details={"endpoint": "test"})
Traceback (most recent call last):
...
kstlib.rapi.exceptions.RapiError: Something went wrong

__init__(self, message: 'str', *, details: 'dict[str, Any] | None' = None) → 'None' -> None[source]¶

Initialize RapiError.

Parameters:

message (str) – Human-readable error message.
details (dict[str, Any] | None) – Additional error context.

exception kstlib.rapi.CredentialError(credential_name, reason)[source]¶

Bases: RapiError

Raised when credential resolution fails.

credential_name¶: Name of the credential that failed.

reason¶: Reason for the failure.

Examples

>>> raise CredentialError("github", "Environment variable not set")
Traceback (most recent call last):
...
kstlib.rapi.exceptions.CredentialError: Credential 'github' failed: Environment variable not set

__init__(self, credential_name: 'str', reason: 'str') → 'None' -> None[source]¶

Initialize CredentialError.

Parameters:

credential_name (str) – Name of the credential that failed.
reason (str) – Reason for the failure.

exception kstlib.rapi.EndpointNotFoundError(endpoint_ref, searched_apis=None)[source]¶

Bases: RapiError

Raised when an endpoint cannot be resolved.

endpoint_ref¶: The endpoint reference that was not found.

searched_apis¶: List of API names that were searched.

Examples

>>> raise EndpointNotFoundError("unknown.endpoint")
Traceback (most recent call last):
...
kstlib.rapi.exceptions.EndpointNotFoundError: Endpoint 'unknown.endpoint' not found

__init__(self, endpoint_ref: 'str', searched_apis: 'list[str] | None' = None) → 'None' -> None[source]¶

Initialize EndpointNotFoundError.

Parameters:

endpoint_ref (str) – The endpoint reference that was not found.
searched_apis (list[str] | None) – List of API names that were searched.

exception kstlib.rapi.EndpointAmbiguousError(endpoint_name, matching_apis)[source]¶

Bases: RapiError

Raised when an endpoint name matches multiple APIs.

endpoint_name¶: The ambiguous endpoint name.

matching_apis¶: List of API names containing this endpoint.

Examples

>>> raise EndpointAmbiguousError("get_data", ["api1", "api2"])
Traceback (most recent call last):
...
kstlib.rapi.exceptions.EndpointAmbiguousError: Endpoint 'get_data' is ambiguous, found in: api1, api2

__init__(self, endpoint_name: 'str', matching_apis: 'list[str]') → 'None' -> None[source]¶

Initialize EndpointAmbiguousError.

Parameters:

endpoint_name (str) – The ambiguous endpoint name.
matching_apis (list[str]) – List of API names containing this endpoint.

exception kstlib.rapi.RequestError(message, *, status_code=None, response_body=None, retryable=False)[source]¶

Bases: RapiError

Raised when an HTTP request fails.

status_code¶: HTTP status code (if available).

response_body¶: Response body (if available).

retryable¶: Whether the error is potentially retryable.

Examples

>>> raise RequestError("Server error", status_code=500, retryable=True)
Traceback (most recent call last):
...
kstlib.rapi.exceptions.RequestError: Server error

__init__(self, message: 'str', *, status_code: 'int | None' = None, response_body: 'str | None' = None, retryable: 'bool' = False) → 'None' -> None[source]¶

Initialize RequestError.

Parameters:

message (str) – Human-readable error message.
status_code (int | None) – HTTP status code (if available).
response_body (str | None) – Response body (if available).
retryable (bool) – Whether the error is potentially retryable.

exception kstlib.rapi.ResponseTooLargeError(response_size, max_size)[source]¶

Bases: RapiError

Raised when response exceeds max_response_size limit.

response_size¶: Actual response size in bytes.

max_size¶: Maximum allowed size in bytes.

Examples

>>> raise ResponseTooLargeError(15_000_000, 10_000_000)
Traceback (most recent call last):
...
kstlib.rapi.exceptions.ResponseTooLargeError: Response size 15000000 exceeds limit 10000000

__init__(self, response_size: 'int', max_size: 'int') → 'None' -> None[source]¶

Initialize ResponseTooLargeError.

Parameters:

response_size (int) – Actual response size in bytes.
max_size (int) – Maximum allowed size in bytes.

exception kstlib.rapi.ConfirmationRequiredError(endpoint_ref, *, expected, actual=None)[source]¶

Bases: RapiError

Raised when a dangerous endpoint requires confirmation.

This exception is raised at runtime when calling an endpoint that has a safeguard configured but the confirm parameter is missing or incorrect.

endpoint_ref¶: Full endpoint reference (api.endpoint).

expected¶: Expected confirmation string.

actual¶: Actual confirmation string provided (None if missing).

Examples

>>> raise ConfirmationRequiredError("api.delete", expected="DELETE X")
Traceback (most recent call last):
...
kstlib.rapi.exceptions.ConfirmationRequiredError: ... requires confirmation...

__init__(self, endpoint_ref: 'str', *, expected: 'str', actual: 'str | None' = None) → 'None' -> None[source]¶

Initialize ConfirmationRequiredError.

Parameters:

endpoint_ref (str) – Full endpoint reference (api.endpoint).
expected (str) – Expected confirmation string.
actual (str | None) – Actual confirmation string provided (None if missing).

exception kstlib.rapi.SafeguardMissingError(endpoint_ref, method)[source]¶

Bases: RapiError

Raised when endpoint requires safeguard but none is configured.

This exception is raised at config load time when an endpoint uses a method that requires a safeguard (e.g., DELETE, PUT) but no safeguard string is provided in the endpoint configuration.

endpoint_ref¶: Full endpoint reference (api.endpoint).

method¶: HTTP method that requires the safeguard.

Examples

>>> raise SafeguardMissingError("api.delete", "DELETE")
Traceback (most recent call last):
...
kstlib.rapi.exceptions.SafeguardMissingError: ... requires a safeguard...

__init__(self, endpoint_ref: 'str', method: 'str') → 'None' -> None[source]¶

Initialize SafeguardMissingError.

Parameters:

endpoint_ref (str) – Full endpoint reference (api.endpoint).
method (str) – HTTP method that requires the safeguard.

exception kstlib.rapi.ServerNotFoundError(server_name, available=None)[source]¶

Bases: RapiError

Raised when a named server profile does not exist.

server_name¶: The server name that was not found.

available¶: List of available server names.

Examples

>>> raise ServerNotFoundError("staging", available=["source", "target"])
Traceback (most recent call last):
...
kstlib.rapi.exceptions.ServerNotFoundError: Server profile 'staging' not found...

__init__(self, server_name: 'str', available: 'list[str] | None' = None) → 'None' -> None[source]¶

Initialize ServerNotFoundError.

Parameters:

server_name (str) – The server name that was not found.
available (list[str] | None) – List of available server names.