swisper-config-service-sdk 0.1.6

Python SDK for the Config Service API - centralized configuration management with hierarchical overrides

Swisper Config Service SDK

Python SDK for the Config Service API - centralized configuration management with hierarchical overrides.

Installation

pip install swisper-config-service-sdk

Or install from source:

pip install ./backend/config-service/sdk

Quick Start

from swisper_config_service_sdk import ConfigServiceClient

async with ConfigServiceClient() as client:
    # Get a configuration value
    config = await client.get("feature-flag")
    print(f"Value: {config.value}")

    # Get with default fallback
    value = await client.get_value("timeout", default="30")

    # Create or update a configuration
    await client.upsert("my-config", "my-value")

Hierarchical Configuration

The Config Service supports three levels of configuration:

1. Global (base) - Default configuration for all agents/nodes
2. Agent-level - Override for a specific agent
3. Node-level - Override for a specific node within an agent

When retrieving a configuration, the service uses fallback: node → agent → global.

async with ConfigServiceClient() as client:
    # Get global config
    global_config = await client.get("feature-flag")

    # Get agent-specific config (falls back to global if not set)
    agent_config = await client.get("feature-flag", agent_id="agent-123")

    # Get node-specific config (falls back to agent, then global)
    node_config = await client.get(
        "feature-flag",
        agent_id="agent-123",
        node_id="node-456"
    )

    # Check where the value came from
    print(f"Source: {node_config.source}")  # "base", "agent", "node", or "cache"

Configuration

Environment Variables

Variable	Default	Description
CONFIG_SERVICE_URL	http://localhost:8090	Base URL of the Config Service
CONFIG_SERVICE_TIMEOUT	30.0	Request timeout in seconds
CONFIG_SERVICE_CONNECT_TIMEOUT	5.0	Connection timeout in seconds
CONFIG_SERVICE_MAX_RETRIES	3	Max retry attempts for transient failures

Programmatic Configuration

from swisper_config_service_sdk import ConfigServiceClient, ConfigServiceSettings

settings = ConfigServiceSettings(
    url="http://config-service.internal:8090",
    timeout=60.0,
    max_retries=5,
)

async with ConfigServiceClient(settings=settings) as client:
    ...

API Reference

ConfigServiceClient

get(name, *, agent_id=None, node_id=None) -> ConfigurationWithSource

Get a configuration by name with hierarchical fallback.

config = await client.get("feature-flag", agent_id="agent-1")
print(config.value)   # The configuration value
print(config.source)  # Where it came from: "base", "agent", "node", "cache"

get_value(name, *, agent_id=None, node_id=None, default=None) -> str | None

Convenience method to get just the value with a default.

value = await client.get_value("timeout", default="30")

upsert(name, value, *, agent_id=None, node_id=None) -> ConfigurationItem

Create or update a configuration.

# Global config
await client.upsert("feature-flag", "enabled")

# Agent-level override
await client.upsert("feature-flag", "disabled", agent_id="agent-1")

# Node-level override
await client.upsert("feature-flag", "beta", agent_id="agent-1", node_id="node-1")

list(*, skip=0, limit=100, agent_id=None, node_id=None) -> list[ConfigurationItem]

List configurations with optional filtering.

# List all configs
configs = await client.list()

# List agent-specific configs
agent_configs = await client.list(agent_id="agent-1")

delete(name) -> bool

Delete a configuration and all its overrides.

await client.delete("old-config")

health() -> HealthStatus

Check service liveness.

status = await client.health()
print(status.status)  # "ok"

ready() -> ReadyStatus

Check service readiness (includes dependency checks).

status = await client.ready()
print(status.redis)  # "connected" or "disconnected"

is_healthy() -> bool

Quick health check returning boolean.

if await client.is_healthy():
    print("Service is healthy")

Error Handling

from swisper_config_service_sdk import (
    ConfigServiceClient,
    ConfigNotFoundError,
    ConfigValidationError,
    ConfigServiceUnavailableError,
    ConfigServiceTimeoutError,
)

async with ConfigServiceClient() as client:
    try:
        config = await client.get("my-config")
    except ConfigNotFoundError as e:
        print(f"Config '{e.name}' not found")
    except ConfigValidationError as e:
        print(f"Invalid request: {e.detail}")
    except ConfigServiceTimeoutError:
        print("Request timed out")
    except ConfigServiceUnavailableError as e:
        print(f"Service unavailable: {e}")

FastAPI Integration

For FastAPI applications, you can use the shared client pattern:

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from typing import Annotated

from swisper_config_service_sdk import (
    ConfigServiceClient,
    connect_shared_client,
    close_shared_client,
    get_shared_client,
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    await connect_shared_client()
    yield
    await close_shared_client()

app = FastAPI(lifespan=lifespan)

async def get_config_client() -> ConfigServiceClient:
    return await get_shared_client()

ConfigClientDep = Annotated[ConfigServiceClient, Depends(get_config_client)]

@app.get("/feature/{name}")
async def get_feature(name: str, client: ConfigClientDep):
    value = await client.get_value(name, default="disabled")
    return {"feature": name, "value": value}

Development

Install dev dependencies

pip install -e ".[dev]"

Run tests

pytest

Run linting

ruff check .
mypy .