Skip to content

replane-dev/replane-python

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Replane Python SDK

Dynamic configuration for Python applications.

Replane Cloud PyPI CI License Community

Replane Screenshot

Replane is a dynamic configuration manager that lets you tweak your software without running scripts or building your own admin panel. Store feature flags, rate limits, UI text, log level, rollout percentage, and more. Delegate editing to teammates and share config across services. No redeploys needed.

Why Dynamic Configuration?

  • Feature flags – toggle features, run A/B tests, roll out to user segments
  • Operational tuning – adjust limits, TTLs, and timeouts without redeploying
  • Per-environment settings – different values for production, staging, dev
  • Incident response – instantly revert to a known-good version
  • Cross-service configuration – share settings with realtime sync
  • Non-engineer access – safe editing with schema validation

Features

  • Real-time updates via Server-Sent Events (SSE)
  • Context-based overrides for feature flags, A/B testing, and gradual rollouts
  • Zero dependencies for sync client (stdlib only)
  • Both sync and async clients available
  • Type-safe with TypedDict support and full type hints
  • Testing utilities with in-memory client

Installation

# Basic installation (sync client only, zero dependencies)
pip install replane

# With async support (adds httpx dependency)
pip install replane[async]

Quick Start

Synchronous Client

from replane import Replane

# Using context manager (recommended)
with Replane(
    base_url="https://cloud.replane.dev",  # or your self-hosted URL
    sdk_key="rp_...",
) as replane:
    # Get a simple config value
    rate_limit = replane.configs["rate_limit"]

    # Get with context for override evaluation
    user_client = replane.with_context({"user_id": user.id, "plan": user.plan})
    feature_enabled = user_client.configs["new_feature"]

    # Get with fallback default
    timeout = replane.configs.get("request_timeout", 30)

Asynchronous Client

Requires pip install replane[async]:

from replane import AsyncReplane

async with AsyncReplane(
    base_url="https://cloud.replane.dev",
    sdk_key="rp_...",
) as replane:
    # Access configs from local cache
    rate_limit = replane.configs["rate_limit"]

    # With context
    enabled = replane.with_context({"plan": "premium"}).configs["feature"]

Without Context Manager

from replane import Replane

# Option 1: Provide credentials in constructor
replane = Replane(
    base_url="https://cloud.replane.dev",
    sdk_key="rp_...",
)
replane.connect()

# Option 2: Provide credentials in connect()
replane = Replane()
replane.connect(
    base_url="https://cloud.replane.dev",
    sdk_key="rp_...",
)

# Use configs
rate_limit = replane.configs["rate_limit"]
user_client = replane.with_context({"user_id": "123"})
feature = user_client.configs["feature_flag"]

# Don't forget to close when done
replane.close()

Type-Safe with Generated Types (Recommended)

Generate TypedDict types from your Replane dashboard for full type safety:

from replane import Replane
from replane_types import Configs  # Generated from Replane dashboard

# Use the Configs TypedDict as a type parameter
with Replane[Configs](
    base_url="https://cloud.replane.dev",
    sdk_key="rp_...",
) as replane:
    # Access configs with dictionary-style notation
    settings = replane.configs["app_settings"]

    # Full type safety - IDE knows the structure of settings
    print(settings["max_upload_size_mb"])
    print(settings["allowed_file_types"])

    # Check if config exists
    if "feature_flag" in replane.configs:
        flag = replane.configs["feature_flag"]

    # Safe access with default
    timeout = replane.configs.get("timeout", 30)

The .configs property provides:

  • Dictionary-style access with replane.configs["config_name"]
  • Type inference when using generated TypedDict types
  • Override evaluation using the default context
  • Familiar dict methods: .get(), .keys(), in operator

Configuration Options

Both clients accept the same configuration:

replane = Replane(
    base_url="https://cloud.replane.dev",
    sdk_key="rp_...",

    # Default context applied to all config evaluations
    context={"environment": "production"},

    # Default values used if server is unavailable during init
    defaults={
        "rate_limit": 100,
        "feature_enabled": False,
    },

    # Configs that must exist (raises error if missing)
    required=["rate_limit", "feature_enabled"],

    # Timeouts in milliseconds
    request_timeout_ms=2000,
    initialization_timeout_ms=5000,
    retry_delay_ms=200,
    inactivity_timeout_ms=30000,

    # Custom agent identifier for User-Agent header
    agent="my-app/1.0.0",

    # Enable debug logging
    debug=True,
)

Context-Based Overrides

Replane evaluates override rules client-side using the context you provide. Your context data never leaves your application.

# Define context based on current user/request
context = {
    "user_id": "user-123",
    "plan": "premium",
    "region": "us-east",
    "is_beta_tester": True,
}

# Overrides are evaluated locally using with_context()
value = replane.with_context(context).configs["feature_flag"]

Scoped Clients with with_context()

Create scoped clients for specific users or requests using with_context():

with Replane(
    base_url="https://cloud.replane.dev",
    sdk_key="rp_...",
) 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"]

    # Can be chained for additional context
    request_client = user_client.with_context({"region": request.region})

The original client is unaffected - scoped clients are lightweight wrappers.

Scoped Defaults with with_defaults()

Create scoped clients with fallback values using with_defaults():

with Replane(
    base_url="https://cloud.replane.dev",
    sdk_key="rp_...",
) as replane:
    # Create a client with fallback defaults
    safe_client = replane.with_defaults({
        "timeout": 30,
        "max_retries": 3,
    })

    # Returns the default if config doesn't exist
    timeout = safe_client.configs["timeout"]  # 30 if not configured

    # Chain with with_context() for both features
    user_client = replane.with_context({"plan": "premium"}).with_defaults({
        "rate_limit": 1000,
    })

Explicit defaults in .configs.get() take precedence over scoped defaults.

Override Examples

Percentage rollout (gradual feature release):

# Server config has 10% rollout based on user_id
# Same user always gets same result (deterministic hashing)
enabled = replane.with_context({"user_id": user.id}).configs["new_checkout"]

Plan-based features:

max_items = replane.with_context({"plan": user.plan}).configs["max_items"]
# Returns different values for free/pro/enterprise plans

Geographic targeting:

content = replane.with_context({"country": request.country}).configs["homepage_banner"]

Subscribing to Changes

React to config changes in real-time:

# Subscribe to all config changes
def on_any_change(name: str, config):
    print(f"Config {name} changed to {config.value}")

unsubscribe = replane.subscribe(on_any_change)

# Subscribe to specific config
def on_feature_change(config):
    update_feature_state(config.value)

unsubscribe_feature = replane.subscribe_config("my_feature", on_feature_change)

# Later: stop receiving updates
unsubscribe()
unsubscribe_feature()

For async clients, callbacks can be async:

async def on_change(name: str, config):
    await notify_services(name, config.value)

replane.subscribe(on_change)

Error Handling

from replane import (
    ReplaneError,
    TimeoutError,
    AuthenticationError,
    NetworkError,
    ErrorCode,
)

try:
    value = replane.configs["my_config"]
except KeyError as e:
    print(f"Config not found: {e}")
except TimeoutError as e:
    print(f"Timed out after {e.timeout_ms}ms")
except AuthenticationError:
    print("Invalid SDK key")
except ReplaneError as e:
    print(f"Error [{e.code}]: {e.message}")

Testing

Use the in-memory client for unit tests:

from replane.testing import create_test_client, InMemoryReplaneClient

# Simple usage
replane = create_test_client({
    "feature_enabled": True,
    "rate_limit": 100,
})

assert replane.configs["feature_enabled"] is True

# With overrides
replane = InMemoryReplaneClient()
replane.set_config(
    "feature",
    value=False,
    overrides=[{
        "name": "premium-users",
        "conditions": [
            {"operator": "in", "property": "plan", "expected": ["pro", "enterprise"]}
        ],
        "value": True,
    }],
)

assert replane.with_context({"plan": "free"}).configs["feature"] is False
assert replane.with_context({"plan": "pro"}).configs["feature"] is True

Pytest Fixture Example

import pytest
from replane.testing import create_test_client

@pytest.fixture
def replane_client():
    return create_test_client({
        "feature_flags": {"dark_mode": True, "new-ui": False},
        "rate_limits": {"default": 100, "premium": 1000},
    })

def test_feature_flag(replane_client):
    flags = replane_client.configs["feature_flags"]
    assert flags["dark_mode"] is True

Manual Lifecycle Management

If you prefer not to use context managers:

# Sync
replane = Replane(base_url="...", sdk_key="...")
replane.connect()  # Blocks until initialized
try:
    value = replane.configs["config"]
finally:
    replane.close()

# Async
replane = AsyncReplane(base_url="...", sdk_key="...")
await replane.connect()
try:
    value = replane.configs["config"]
finally:
    await replane.close()

Framework Integration

FastAPI

from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import FastAPI, Depends
from replane import AsyncReplane

_replane: AsyncReplane | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global _replane
    _replane = AsyncReplane(
        base_url="https://cloud.replane.dev",
        sdk_key="rp_...",
    )
    await _replane.connect()
    yield
    await _replane.close()

app = FastAPI(lifespan=lifespan)

def get_replane() -> AsyncReplane:
    assert _replane is not None
    return _replane

# Define reusable dependency type
Replane = Annotated[AsyncReplane, Depends(get_replane)]

@app.get("/items")
async def get_items(replane: Replane):
    max_items = replane.with_context({"plan": "free"}).configs["max_items"]
    return {"max_items": max_items}

Flask

from flask import Flask, g
from replane import Replane

app = Flask(__name__)
_replane: Replane | None = None

@app.before_first_request
def init_replane():
    global _replane
    _replane = Replane(
        base_url="https://cloud.replane.dev",
        sdk_key="rp_...",
    )
    _replane.connect()

@app.route("/items")
def get_items():
    max_items = _replane.configs["max_items"]
    return {"max_items": max_items}

Requirements

  • Python 3.10+
  • No dependencies for sync client
  • httpx for async client (pip install replane[async])

Contributing

See CONTRIBUTING.md for development setup and contribution guidelines.

Community

Have questions or want to discuss Replane? Join the conversation in GitHub Discussions.

License

MIT