Python SDK - API Reference

Every public symbol exported by westyx_nexus. Both NexusClient (sync) and AsyncNexusClient (async) share the same read API - only the lifecycle methods (sync, connect_stream, close) differ.

Construction

`NexusClient.create(config: NexusConfig) -> NexusClient`

Synchronous factory. Performs a blocking initial sync.

nexus = NexusClient.create(NexusConfig(base_url="...", api_key="..."))

`await AsyncNexusClient.create(config: NexusConfig) -> AsyncNexusClient`

Async factory. Same blocking initial sync semantics, but expressed as a coroutine.

nexus = await AsyncNexusClient.create(NexusConfig(base_url="...", api_key="..."))

Raises	When
`NexusInitError`	Initial sync failed for any reason. The `__cause__` attribute carries the underlying exception.
`ValueError`	`base_url` or `api_key` missing or invalid.

Configs

`nexus.get_config(key: str, default: Any = None) -> Any`

Returns the resolved config value as Any (the wire-decoded JSON value), or default if the key is absent.

db_url   = nexus.get_config("database.url")                  # str | None
fallback = nexus.get_config("missing", default="localhost")  # str

`nexus.get_config_as(key: str, type_: type[T], default: T | None = None) -> T | None`

Returns the config value coerced to the given type, using a standard call (type_(value)).

port = nexus.get_config_as("server.port", int, default=8080)   # int | None
host = nexus.get_config_as("server.host", str, default="0.0.0.0")

Raises	When
`NexusError`	The cached value cannot be coerced (e.g. asking for `int` on a non-numeric string).

`nexus.get_all_configs() -> dict[str, Any]`

Snapshot copy of every config key-value pair currently in the cache.

Feature flags

`nexus.get_flag(key: str, default: bool = False) -> bool`

Returns the is_active state of a flag, or default when the key is absent.

nexus.get_flag("checkout.v2")             # False if missing
nexus.get_flag("checkout.v2", True)       # True if missing

`nexus.get_all_flags() -> dict[str, bool]`

Snapshot copy of every flag key-is_active pair in the cache.

AB Testing (v0.3.0)

`nexus.evaluate_ab(keys: list[str], user_id: str, attributes: dict[str, Any]) -> dict[str, bool]`

Batch-evaluate AB Testing flags for a given user. Issues a single POST /v1/flags/evaluate-ab request; the server applies each flag’s rollout / targeting rules to the supplied user_id and attributes, and returns a key -> bool map. Unlike get_flag, this is a network call - the result depends on user_id, so it cannot be served from the local snapshot cache.

results = nexus.evaluate_ab(
    keys=["checkout.v2", "homepage.banner"],
    user_id="user-42",
    attributes={"country": "DE", "plan": "pro"},
)
if results.get("checkout.v2", False):
    render_new_checkout()

results = await async_nexus.evaluate_ab(
    keys=["checkout.v2", "homepage.banner"],
    user_id="user-42",
    attributes={"country": "DE", "plan": "pro"},
)

Raises	When
`NexusAbAddonNotAvailableError`	Server returned HTTP 403 - the AB Testing add-on is not active
`NexusUnauthorizedError`	Server returned HTTP 401
`NexusBillingError`	Server returned HTTP 402 - tenant invoices are overdue
`NexusNotFoundError`	Server returned HTTP 404

Secrets (secret key only)

`nexus.get_secret(key: str) -> str`

Returns the plaintext value of a text-type secret.

password = nexus.get_secret("DB_PASSWORD")

Raises	When
`NexusPublicKeyError`	The client uses a public key
`NexusServiceKindMismatchError`	The service `kind` reported by the server is `"frontend"`
`NexusSecretNotFoundError`	The key is not in the cache (the `key` attribute carries the requested name)

`nexus.get_secret_file_path(key: str) -> str | None` (v0.3.0)

Returns the on-disk path of a file-type secret, or None. The SDK materialises the value as a temp file on every sync and exposes the absolute path here:

tempfile.gettempdir()/nexus-<key>-<sha256-prefix>

# Server stores GOOGLE_APPLICATION_CREDENTIALS as a file-type secret.
gcp_creds_path = nexus.get_secret_file_path("GOOGLE_APPLICATION_CREDENTIALS")
if gcp_creds_path is not None:
    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = gcp_creds_path

Returns None for text-type secrets, unknown keys, or when the on-disk write failed - fall back to get_secret for text-type values.

Secrets - Write API (v0.5.0) (secret key only)

`nexus.set_secret(key: str, value: str, secret_type: str = "text") -> None`

Creates or updates a secret with the given key and value.

nexus.set_secret("DB_PASSWORD", "new-password")
nexus.set_secret("TLS_CERT", cert_pem, secret_type="file")

Raises	When
`NexusPublicKeyError`	The client uses a public key
`NexusRateLimitedError`	Server returned HTTP 429 - rate limit exceeded on write endpoints
`NexusUnauthorizedError`	Server returned HTTP 401
`NexusServiceKindMismatchError`	The service `kind` is `"frontend"`

`nexus.delete_secret(key: str) -> None`

Deletes a secret by key.

nexus.delete_secret("DEPRECATED_API_KEY")

Raises	When
`NexusPublicKeyError`	The client uses a public key
`NexusRateLimitedError`	Server returned HTTP 429 - rate limit exceeded on write endpoints
`NexusUnauthorizedError`	Server returned HTTP 401
`NexusServiceKindMismatchError`	The service `kind` is `"frontend"`

`nexus.delete_secret_version(key: str, version: int) -> None` (v0.5.0)

Deletes a specific version of a secret.

nexus.delete_secret_version("DB_PASSWORD", version=3)

Raises	When
`NexusPublicKeyError`	The client uses a public key
`NexusRateLimitedError`	Server returned HTTP 429 - rate limit exceeded on write endpoints
`NexusUnauthorizedError`	Server returned HTTP 401
`NexusServiceKindMismatchError`	The service `kind` is `"frontend"`

Metadata

Property	Type	Description
`nexus.key_type`	`str`	`"sk"` or `"pk"` - derived from the configured `api_key`
`nexus.synced_at`	`float`	Unix timestamp of the last successful sync
`nexus.stream_connected`	`bool`	`True` while the SSE stream is connected and pushing events

Sync / lifecycle (sync API)

`nexus.sync() -> None`

Forces a sync now, bypassing the TTL. Rarely needed - the SDK syncs automatically.

`nexus.connect_stream() -> None` / `nexus.disconnect_stream() -> None`

Start/stop the SSE thread. The SDK calls connect_stream() automatically after the initial sync (unless stream=False in config). Idempotent.

`nexus.close() -> None`

Stops the SSE thread and releases the underlying HTTP client. Always call on shutdown (or use a with block).

Sync / lifecycle (async API)

The async client has matching coroutine versions:

await nexus.sync()
nexus.connect_stream()             # not async - just spawns a task
await nexus.disconnect_stream()    # awaits the task to fully stop
await nexus.close()                # awaits stream shutdown + httpx aclose()

Use async with await AsyncNexusClient.create(config) as nexus: for guaranteed cleanup.

Error types

from westyx_nexus import (
    NexusError,                       # base class
    NexusInitError,                   # initial sync failure (cause carries detail)
    NexusUnauthorizedError,           # HTTP 401
    NexusBillingError,                # HTTP 402 (v0.2.0)
    NexusNotFoundError,               # HTTP 404
    NexusPublicKeyError,              # get_secret with public key
    NexusServiceKindMismatchError,    # get_secret on frontend service (v0.2.0)
    NexusSecretNotFoundError,         # unknown secret key (key attribute on instance)
    NexusAbAddonNotAvailableError,    # evaluate_ab - HTTP 403 (v0.3.0)
)

Logger interface

from typing import Protocol

class NexusLogger(Protocol):
    def debug(self, message: str) -> None: ...
    def info(self, message: str) -> None: ...
    def error(self, message: str, cause: BaseException | None = None) -> None: ...

A duck-typed Protocol - any object with these three methods qualifies. The SDK ships a StdlibNexusLogger adapter that forwards to the stdlib logging module.

Overview

Backend SDKs

Browser SDKs

OpenFeature Providers

Python SDK - API Reference

Construction

`NexusClient.create(config: NexusConfig) -> NexusClient`

`await AsyncNexusClient.create(config: NexusConfig) -> AsyncNexusClient`

Configs

`nexus.get_config(key: str, default: Any = None) -> Any`

`nexus.get_config_as(key: str, type_: type[T], default: T | None = None) -> T | None`

`nexus.get_all_configs() -> dict[str, Any]`

Feature flags

`nexus.get_flag(key: str, default: bool = False) -> bool`

`nexus.get_all_flags() -> dict[str, bool]`

AB Testing (v0.3.0)

`nexus.evaluate_ab(keys: list[str], user_id: str, attributes: dict[str, Any]) -> dict[str, bool]`

Secrets (secret key only)

`nexus.get_secret(key: str) -> str`

`nexus.get_secret_file_path(key: str) -> str | None` (v0.3.0)

Secrets - Write API (v0.5.0) (secret key only)

`nexus.set_secret(key: str, value: str, secret_type: str = "text") -> None`

`nexus.delete_secret(key: str) -> None`

`nexus.delete_secret_version(key: str, version: int) -> None` (v0.5.0)

Metadata

Sync / lifecycle (sync API)

`nexus.sync() -> None`

`nexus.connect_stream() -> None` / `nexus.disconnect_stream() -> None`

`nexus.close() -> None`

Sync / lifecycle (async API)

Error types

Logger interface

​Construction

​NexusClient.create(config: NexusConfig) -> NexusClient

​await AsyncNexusClient.create(config: NexusConfig) -> AsyncNexusClient

​Configs

​nexus.get_config(key: str, default: Any = None) -> Any

​nexus.get_config_as(key: str, type_: type[T], default: T | None = None) -> T | None

​nexus.get_all_configs() -> dict[str, Any]

​Feature flags

​nexus.get_flag(key: str, default: bool = False) -> bool

​nexus.get_all_flags() -> dict[str, bool]

​AB Testing (v0.3.0)

​nexus.evaluate_ab(keys: list[str], user_id: str, attributes: dict[str, Any]) -> dict[str, bool]

​Secrets (secret key only)

​nexus.get_secret(key: str) -> str

​nexus.get_secret_file_path(key: str) -> str | None (v0.3.0)

​Secrets - Write API (v0.5.0) (secret key only)

​nexus.set_secret(key: str, value: str, secret_type: str = "text") -> None

​nexus.delete_secret(key: str) -> None

​nexus.delete_secret_version(key: str, version: int) -> None (v0.5.0)

​Metadata

​Sync / lifecycle (sync API)

​nexus.sync() -> None

​nexus.connect_stream() -> None / nexus.disconnect_stream() -> None

​nexus.close() -> None

​Sync / lifecycle (async API)

​Error types

​Logger interface

Construction

`NexusClient.create(config: NexusConfig) -> NexusClient`

`await AsyncNexusClient.create(config: NexusConfig) -> AsyncNexusClient`

Configs

`nexus.get_config(key: str, default: Any = None) -> Any`

`nexus.get_config_as(key: str, type_: type[T], default: T | None = None) -> T | None`

`nexus.get_all_configs() -> dict[str, Any]`

Feature flags

`nexus.get_flag(key: str, default: bool = False) -> bool`

`nexus.get_all_flags() -> dict[str, bool]`

AB Testing (v0.3.0)

`nexus.evaluate_ab(keys: list[str], user_id: str, attributes: dict[str, Any]) -> dict[str, bool]`

Secrets (secret key only)

`nexus.get_secret(key: str) -> str`

`nexus.get_secret_file_path(key: str) -> str | None` (v0.3.0)

Secrets - Write API (v0.5.0) (secret key only)

`nexus.set_secret(key: str, value: str, secret_type: str = "text") -> None`

`nexus.delete_secret(key: str) -> None`

`nexus.delete_secret_version(key: str, version: int) -> None` (v0.5.0)

Metadata

Sync / lifecycle (sync API)

`nexus.sync() -> None`

`nexus.connect_stream() -> None` / `nexus.disconnect_stream() -> None`

`nexus.close() -> None`

Sync / lifecycle (async API)

Error types

Logger interface