API Reference¶

This page provides detailed API documentation for the Replane Python SDK.

Clients¶

Replane¶

class replane.Replane(base_url=None, sdk_key=None, *, context=None, defaults=None, required=None, request_timeout_ms=2000, initialization_timeout_ms=5000, retry_delay_ms=200, inactivity_timeout_ms=30000, agent=None, debug=False)[source]¶

Bases: Generic[ConfigsT]

Synchronous Replane client with background SSE streaming.

This client maintains a persistent SSE connection to receive real-time config updates. Config reads are synchronous and return immediately from the local cache.

The optional type parameter ConfigsT can be a TypedDict generated by Replane’s codegen feature. When provided, it enables better type inference for config values accessed via the configs property.

Example with context manager (recommended):

>>> with Replane(
...     base_url="https://cloud.replane.dev",
...     sdk_key="rp_...",
... ) as replane:
...     value = replane.configs["feature-flag"]

Without context manager (credentials in constructor):

>>> replane = Replane(base_url="...", sdk_key="...")
>>> replane.connect()
>>> value = replane.configs["feature-flag"]
>>> replane.close()

Without context manager (credentials in connect):

>>> replane = Replane()
>>> replane.connect(base_url="...", sdk_key="...")
>>> value = replane.configs["feature-flag"]
>>> replane.close()

With generated types for better type safety (recommended):

>>> from replane_types import Configs
>>> with Replane[Configs](...) as replane:
...     config = replane.configs["feature-flag"]  # fully typed
...     print(config["enabled"])  # type-safe property access

close()[source]¶

Close the client and stop the SSE connection.

Return type:: None

connect(base_url=None, sdk_key=None, *, wait=True)[source]¶

Connect to the Replane server and start receiving updates.

This starts a background thread that maintains the SSE connection.

Parameters:

base_url (str | None, default: None) – Base URL of the Replane server. Overrides constructor value.
sdk_key (str | None, default: None) – SDK key for authentication. Overrides constructor value.
wait (bool, default: True) – If True, block until initial configs are loaded.

Raises:

ValueError – If base_url or sdk_key is not provided here or in constructor.
ReplaneError – If connection fails or required configs are missing.

Return type:

None

is_initialized()[source]¶

Check if the client has finished initialization.

Return type:: bool
Returns:: True if the client has received initial configs from the server.

subscribe(callback)[source]¶

Subscribe to all config changes.

Parameters:: callback (Callable[[str, Config], None]) – Function called with (config_name, config) on changes.
Return type:: Callable[[], None]
Returns:: Unsubscribe function.

subscribe_config(name, callback)[source]¶

Subscribe to changes for a specific config.

Parameters:

name (str) – Config name to watch.
callback (Callable[[Config], None]) – Function called with the new config on changes.

Return type:

Callable[[], None]

Returns:

Unsubscribe function.

wait_for_init()[source]¶

Wait for the client to finish initialization.

Raises:

TimeoutError – If initialization takes too long.
ReplaneError – If initialization fails.

Return type:

None

with_context(context)[source]¶

Create a contextual wrapper with additional context.

Returns a new object that wraps this client and uses the merged context for all operations. The original client is unaffected.

This is useful for creating scoped clients for specific users or requests:

Example

>>> with Replane(...) as replane:
...     # Create a scoped client for a specific user
...     user_client = replane.with_context({
...         "user_id": user.id,
...         "plan": user.plan,
...     })
...     # All operations use the merged context
...     rate_limit = user_client.configs["rate-limit"]
...     settings = user_client.configs["app-settings"]

Parameters:: context (dict[str, str | int | float | bool | None]) – Additional context to merge with the client’s default context.
Return type:: ContextualReplane[TypeVar(ConfigsT)]
Returns:: A ContextualReplane wrapper with the merged context.

with_defaults(defaults)[source]¶

Create a contextual wrapper with additional defaults.

Returns a new object that wraps this client and uses the merged defaults for all operations. The original client is unaffected.

This is useful for providing fallback values for specific use cases:

Example

>>> with Replane(...) as replane:
...     # Create a scoped client with additional defaults
...     safe_client = replane.with_defaults({
...         "timeout": 30,
...         "max-retries": 3,
...     })
...     # Returns 30 if "timeout" is not configured
...     timeout = safe_client.configs["timeout"]

Parameters:: defaults (dict[str, Any]) – Additional defaults to use when configs are not found.
Return type:: ContextualReplane[TypeVar(ConfigsT)]
Returns:: A ContextualReplane wrapper with the additional defaults.

property configs: ConfigsT¶

Dictionary-like accessor for configuration values.

Provides bracket notation access to configs with override evaluation. When using generated TypedDict types, provides full type safety.

Example

>>> config = replane.configs["my-feature-flag"]
>>> print(config["enabled"])

Returns:: A ConfigAccessor that behaves like a typed dictionary.

AsyncReplane¶

class replane.AsyncReplane(base_url=None, sdk_key=None, *, context=None, defaults=None, required=None, request_timeout_ms=2000, initialization_timeout_ms=5000, retry_delay_ms=200, inactivity_timeout_ms=30000, agent=None, debug=False)[source]¶

Bases: Generic[ConfigsT]

Asynchronous Replane client with background SSE streaming.

This client maintains a persistent SSE connection to receive real-time config updates. All operations are async and non-blocking.

Requires httpx: pip install replane[async]

Example with context manager (recommended):

>>> async with AsyncReplane(
...     base_url="https://cloud.replane.dev",
...     sdk_key="rp_...",
... ) as replane:
...     value = replane.configs["feature-flag"]

Without context manager (credentials in constructor):

>>> replane = AsyncReplane(base_url="...", sdk_key="...")
>>> await replane.connect()
>>> value = replane.configs["feature-flag"]
>>> await replane.close()

Without context manager (credentials in connect):

>>> replane = AsyncReplane()
>>> await replane.connect(base_url="...", sdk_key="...")
>>> value = replane.configs["feature-flag"]
>>> await replane.close()

With generated types for better type safety (recommended):

>>> from replane_types import Configs
>>> async with AsyncReplane[Configs](...) as replane:
...     config = replane.configs["feature-flag"]  # fully typed
...     print(config["enabled"])  # type-safe property access

async close()[source]¶

Close the client and stop the SSE connection.

Return type:: None

async connect(base_url=None, sdk_key=None, *, wait=True)[source]¶

Connect to the Replane server and start receiving updates.

This starts a background task that maintains the SSE connection.

Parameters:

base_url (str | None, default: None) – Base URL of the Replane server. Overrides constructor value.
sdk_key (str | None, default: None) – SDK key for authentication. Overrides constructor value.
wait (bool, default: True) – If True, wait until initial configs are loaded.

Raises:

ValueError – If base_url or sdk_key is not provided here or in constructor.
ReplaneError – If connection fails or required configs are missing.

Return type:

None

is_initialized()[source]¶

Check if the client has finished initialization.

Return type:: bool
Returns:: True if the client has received initial configs from the server.

subscribe(callback)[source]¶

Subscribe to all config changes.

The callback can be sync or async.

Parameters:: callback (Callable[[str, Config], Optional[Awaitable[None]]]) – Function called with (config_name, config) on changes.
Return type:: Callable[[], None]
Returns:: Unsubscribe function.

subscribe_config(name, callback)[source]¶

Subscribe to changes for a specific config.

The callback can be sync or async.

Parameters:

name (str) – Config name to watch.
callback (Callable[[Config], Optional[Awaitable[None]]]) – Function called with the new config on changes.

Return type:

Callable[[], None]

Returns:

Unsubscribe function.

async wait_for_init()[source]¶

Wait for the client to finish initialization.

Raises:

TimeoutError – If initialization takes too long.
ReplaneError – If initialization fails.

Return type:

None

with_context(context)[source]¶

Create a contextual wrapper with additional context.

Returns a new object that wraps this client and uses the merged context for all operations. The original client is unaffected.

This is useful for creating scoped clients for specific users or requests:

Example

>>> async with AsyncReplane(...) as replane:
...     # Create a scoped client for a specific user
...     user_client = replane.with_context({
...         "user_id": user.id,
...         "plan": user.plan,
...     })
...     # All operations use the merged context
...     rate_limit = user_client.configs["rate-limit"]
...     settings = user_client.configs["app-settings"]

Parameters:: context (dict[str, str | int | float | bool | None]) – Additional context to merge with the client’s default context.
Return type:: ContextualAsyncReplane[TypeVar(ConfigsT)]
Returns:: A ContextualAsyncReplane wrapper with the merged context.

with_defaults(defaults)[source]¶

Create a contextual wrapper with additional defaults.

Returns a new object that wraps this client and uses the merged defaults for all operations. The original client is unaffected.

This is useful for providing fallback values for specific use cases:

Example

>>> async with AsyncReplane(...) as replane:
...     # Create a scoped client with additional defaults
...     safe_client = replane.with_defaults({
...         "timeout": 30,
...         "max-retries": 3,
...     })
...     # Returns 30 if "timeout" is not configured
...     timeout = safe_client.configs["timeout"]

Parameters:: defaults (dict[str, Any]) – Additional defaults to use when configs are not found.
Return type:: ContextualAsyncReplane[TypeVar(ConfigsT)]
Returns:: A ContextualAsyncReplane wrapper with the additional defaults.

property configs: ConfigsT¶

Dictionary-like accessor for configuration values.

Provides bracket notation access to configs with override evaluation. When using generated TypedDict types, provides full type safety.

Note: Despite being on the async client, this accessor is synchronous since it only reads from the local cache without any I/O.

Example

>>> config = replane.configs["my-feature-flag"]
>>> print(config["enabled"])

Returns:: An AsyncConfigAccessor that behaves like a typed dictionary.

Testing¶

InMemoryReplaneClient¶

class replane.testing.InMemoryReplaneClient(initial_configs=None, *, context=None)[source]¶

Bases: Generic[ConfigsT]

An in-memory Replane client for testing.

This client provides the same interface as Replane but stores all configs in memory. It’s useful for unit tests where you don’t want to connect to a real Replane server.

Example

>>> replane = InMemoryReplaneClient({
...     "feature-enabled": True,
...     "rate-limit": 100,
... })
>>> assert replane.configs["feature-enabled"] is True
>>> assert replane.configs["rate-limit"] == 100

With overrides:

>>> replane = InMemoryReplaneClient()
>>> replane.set_config(
...     "feature",
...     value=False,
...     overrides=[{
...         "name": "beta",
...         "conditions": [{"operator": "equals", "property": "plan", "expected": "beta"}],
...         "value": True,
...     }],
... )
>>> assert replane.with_context({"plan": "free"}).configs["feature"] is False
>>> assert replane.with_context({"plan": "beta"}).configs["feature"] is True

close()[source]¶

Close the client.

Return type:: None

delete(name)[source]¶

Delete a config.

Parameters:: name (str) – Config name to delete.
Return type:: bool
Returns:: True if config was deleted, False if it didn’t exist.

is_initialized()[source]¶

Check if the client has finished initialization.

For the in-memory client, this always returns True since configs are available immediately.

Return type:: bool
Returns:: True (always, for in-memory client).

set(name, value)[source]¶

Set a config value (simple form without overrides).

Parameters:

name (str) – Config name.
value (Any) – Config value.

Return type:

None

set_config(name, value, *, overrides=None)[source]¶

Set a config with optional overrides.

Parameters:

name (str) – Config name.
value (Any) – Base config value.
overrides (list[dict[str, Any]] | None, default: None) – Optional list of override rules.

Return type:

None

Example

>>> client.set_config(
...     "rate-limit",
...     value=100,
...     overrides=[{
...         "name": "premium-users",
...         "conditions": [
...             {"operator": "in", "property": "plan", "expected": ["pro", "enterprise"]}
...         ],
...         "value": 1000,
...     }],
... )

subscribe(callback)[source]¶

Subscribe to all config changes.

Parameters:: callback (Callable[[str, Config], None]) – Function called with (config_name, config) on changes.
Return type:: Callable[[], None]
Returns:: Unsubscribe function.

subscribe_config(name, callback)[source]¶

Subscribe to changes for a specific config.

Parameters:

name (str) – Config name to watch.
callback (Callable[[Config], None]) – Function called with the new config on changes.

Return type:

Callable[[], None]

Returns:

Unsubscribe function.

with_context(context)[source]¶

Create a scoped client with additional context.

Return type:: ContextualInMemoryClient[TypeVar(ConfigsT)]

with_defaults(defaults)[source]¶

Create a scoped client with additional defaults.

Return type:: ContextualInMemoryClient[TypeVar(ConfigsT)]

property configs: ConfigsT¶: Dictionary-like accessor for configs.

create_test_client¶

replane.testing.create_test_client(configs=None, *, context=None)[source]¶

Create an in-memory client for testing.

This is a convenience function for creating test clients.

Parameters:

configs (dict[str, Any] | None, default: None) – Optional dict of config name -> value.
context (dict[str, str | int | float | bool | None] | None, default: None) – Default context for override evaluation.

Return type:

InMemoryReplaneClient

Returns:

An InMemoryReplaneClient instance.

Example

>>> client = create_test_client({
...     "feature-enabled": True,
...     "max-items": 50,
... })

Types¶

Config¶

class replane.Config(name, value, overrides=<factory>)[source]

A configuration with its base value and override rules.

name: Unique identifier for this config.

value: The base/default value returned when no overrides match.

overrides: List of override rules evaluated in order.

Attributes:

name (str): Unique identifier for this config.
value (Any): The base/default value returned when no overrides match.
overrides (tuple[Override, …]): List of override rules evaluated in order.

Override¶

class replane.Override(name, conditions, value)[source]: An override rule that returns a specific value when conditions match.

Attributes:

name (str): Name/description of this override rule.
conditions (tuple[Condition, …]): Conditions that must all match.
value (Any): Value to return when conditions match.

Conditions¶

class replane.PropertyCondition(operator, property, expected)[source]: A condition that compares a context property against expected values.

class replane.SegmentationCondition(operator, property, from_percentage, to_percentage, seed)[source]: A condition for percentage-based bucketing (gradual rollouts).

class replane.AndCondition(operator, conditions)[source]: Logical AND of multiple conditions.

class replane.OrCondition(operator, conditions)[source]: Logical OR of multiple conditions.

class replane.NotCondition(operator, condition)[source]: Logical NOT of a condition.

Context¶

Context is a TypedDict representing runtime context for override evaluation. It’s a dictionary with string keys and primitive values (str, int, float, bool, or None).

from replane import Context

context: Context = {
    "user_id": "user-123",
    "plan": "premium",
    "region": "us-east",
    "is_beta": True,
}

Errors¶

ReplaneError¶

class replane.ReplaneError(code, message, *, cause=None)[source]¶

Bases: Exception

Base exception for all Replane SDK errors.

code¶: Error code identifying the type of error.

message¶: Human-readable error description.

cause¶: Optional underlying exception that caused this error.

Example

try:

value = client.get(“my-config”)

except ReplaneError as e:

if e.code == ErrorCode.NOT_FOUND:: # Handle missing config pass
elif e.code == ErrorCode.TIMEOUT:: # Handle timeout pass

ErrorCode¶

class replane.ErrorCode(*values)[source]¶

Error codes for ReplaneError.

AUTH_ERROR = 'auth_error'¶

CLIENT_ERROR = 'client_error'¶

CLOSED = 'closed'¶

FORBIDDEN = 'forbidden'¶

MISSING_DEPENDENCY = 'missing_dependency'¶

NETWORK_ERROR = 'network_error'¶

NOT_FOUND = 'not_found'¶

NOT_INITIALIZED = 'not_initialized'¶

SERVER_ERROR = 'server_error'¶

TIMEOUT = 'timeout'¶

UNKNOWN = 'unknown'¶

Specific Errors¶

class replane.ConfigNotFoundError(config_name, *, cause=None)[source]¶

Bases: ReplaneError

Raised when a requested config does not exist.

class replane.TimeoutError(message='Operation timed out', *, timeout_ms=None, cause=None)[source]¶

Bases: ReplaneError

Raised when an operation times out.

class replane.AuthenticationError(message='Authentication failed - check your SDK key', *, cause=None)[source]¶

Bases: ReplaneError

Raised when authentication fails (invalid SDK key).

class replane.NetworkError(message='Network request failed', *, cause=None)[source]¶

Bases: ReplaneError

Raised when a network request fails.

class replane.ClientClosedError(*, cause=None)[source]¶

Bases: ReplaneError

Raised when attempting operations on a closed client.

class replane.NotInitializedError(*, cause=None)[source]¶

Bases: ReplaneError

Raised when the client hasn’t finished initialization.

class replane.MissingDependencyError(dependency, feature)[source]¶

Bases: ReplaneError

Raised when an optional dependency is required but not installed.

Version¶

replane.VERSION = '1.0.0'¶

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to ‘utf-8’. errors defaults to ‘strict’.

replane.VERSION_SHORT = '1.0'¶

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str