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` missing, not `https` (loopback excepted), auth missing (no `api_key` and no enabled `wif`), or `sse_reconnect_cooldown` not an `int` / non-empty `list[int]`.

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.find_flag(key: str) -> bool | None`

Returns the flag’s is_active state, or None when the flag is not in the cache at all. get_flag answers False both for a flag that is off and for one that does not exist. Anything that has to tell those apart - an OpenFeature provider answering FLAG_NOT_FOUND, or a startup check that a flag key is spelled the way the console spells it - needs find_flag. get_flag is expressed in terms of it, so the two cannot drift.

nexus.find_flag("checkout.v2")      # True / False / None

Available on AsyncNexusClient with the same signature.

`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. Attribute values must be strings. Cohort conditions compare them as strings with eq / neq / in, so a non-string value raises TypeError naming the attribute and the type that arrived, before the request is sent. Nothing is stringified: no conversion is lossless, and a value that silently matches no rule is harder to diagnose than one that was refused. A None value is treated as unset and dropped rather than kept as "", which could satisfy an eq "" rule.

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 file in a private per-client directory and exposes the absolute path here:

<per-client 0700 directory under tempfile.gettempdir()>/<sha256(key)>-<sha256(value)>

Both the key and the value are hashed - the secret’s name never appears on the filesystem. The directory scopes every path to one client, so two clients holding the same secret do not share a file. The file is created with mode 0600. See File-type secret security for the full guarantees and lifecycle.

# 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)

Since v0.15.0 these three methods are available on AsyncNexusClient too, with the same signatures, as coroutines: await async_nexus.set_secret(...), await async_nexus.delete_secret(...), await async_nexus.delete_secret_version(...).

`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
`NexusBillingError`	Server returned HTTP 402 - tenant invoices are overdue
`NexusNotFoundError`	Server returned HTTP 404 - project or service not found

`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
`NexusBillingError`	Server returned HTTP 402 - tenant invoices are overdue
`NexusNotFoundError`	Server returned HTTP 404 - key does not exist

`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
`NexusBillingError`	Server returned HTTP 402 - tenant invoices are overdue
`NexusNotFoundError`	Server returned HTTP 404 - key or version does not exist

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

All fourteen are exported from the package root:

from westyx_nexus import (
    NexusError,                       # base class - catch this to catch everything
    NexusInitError,                   # initial sync failure (__cause__ carries detail)
    NexusUnauthorizedError,           # HTTP 401
    NexusBillingError,                # HTTP 402 (v0.2.0)
    NexusAbAddonNotAvailableError,    # evaluate_ab - HTTP 403 (v0.3.0)
    NexusNotFoundError,               # HTTP 404
    NexusRateLimitedError,            # HTTP 429 on a write endpoint
    NexusQuarantinedError,            # HTTP 429 with a quarantine body (v0.4.0)
    NexusPublicKeyError,              # secret access with a public key
    NexusServiceKindMismatchError,    # secret access on a frontend service (v0.2.0)
    NexusSecretNotFoundError,         # unknown secret key (key attribute on instance)
    NexusSessionExpiredError,         # WIF session JWT expired
    NexusWIFNotConfiguredError,       # WIF selected but not usable in this environment
    NexusWIFTokenExchangeFailedError, # WIF token exchange rejected or unavailable
)

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.find_flag(key: str) -> bool | None`

`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.find_flag(key: str) -> bool | None

​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.find_flag(key: str) -> bool | None`

`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