Python SDK
Flipswitch SDK for Python with real-time SSE support
The Flipswitch Python SDK provides an OpenFeature-compatible provider with automatic cache invalidation via Server-Sent Events (SSE).
GitHub: github.com/flipswitch-io/python-sdk
Requirements
- Python 3.9+
- OpenFeature SDK
Installation
pip install flipswitch-sdk openfeature-sdkpoetry add flipswitch-sdk openfeature-sdkQuick Start
from flipswitch import FlipswitchProvider
from openfeature import api
from openfeature.evaluation_context import EvaluationContext
# Only API key is required
provider = FlipswitchProvider(api_key="YOUR_API_KEY")
# Register with OpenFeature
api.set_provider(provider)
# Get a client and evaluate flags
client = api.get_client()
dark_mode = client.get_boolean_value("dark-mode", False)
welcome_message = client.get_string_value("welcome-message", "Hello!")
max_items = client.get_integer_value("max-items-per-page", 10)Configuration Options
provider = FlipswitchProvider(
api_key="YOUR_API_KEY",
base_url="https://api.flipswitch.io", # Optional: defaults to production
enable_realtime=True, # Optional: defaults to True
http_client=custom_httpx_client, # Optional: custom httpx.Client
enable_polling_fallback=True, # Optional: fall back to polling when SSE fails
polling_interval=30.0, # Optional: polling interval in seconds
max_sse_retries=5, # Optional: max SSE retries before fallback
)| Option | Type | Default | Description |
|---|---|---|---|
api_key | str | required | Environment API key from dashboard |
base_url | str | https://api.flipswitch.io | Your Flipswitch server URL |
enable_realtime | bool | True | Enable SSE for real-time flag updates |
http_client | httpx.Client | None | Custom HTTP client |
enable_polling_fallback | bool | True | Fall back to polling when SSE fails |
polling_interval | float | 30.0 | Polling interval in seconds |
max_sse_retries | int | 5 | Max SSE retries before polling fallback |
Evaluation Context
Pass user attributes for targeting:
context = EvaluationContext(
targeting_key="user-123",
attributes={
"email": "user@example.com",
"plan": "premium",
"country": "SE",
},
)
show_feature = client.get_boolean_value("new-feature", False, context)Real-Time Updates
When enable_realtime=True is set (default), the SDK maintains an SSE connection to receive instant flag changes:
- The SSE client receives a
flag-updatedorconfig-updatedevent - The local cache is immediately invalidated
- Next flag evaluation fetches the fresh value
Event Listeners
from flipswitch.types import FlagChangeEvent, ConnectionStatus
def on_flag_change(event: FlagChangeEvent):
if event.flag_key:
print(f"Flag changed: {event.flag_key}")
else:
print("All flags invalidated")
provider.add_flag_change_listener(on_flag_change)
# Remove listener when done
provider.remove_flag_change_listener(on_flag_change)Connection Status
from flipswitch.types import ConnectionStatus
# Check current SSE status
status = provider.get_sse_status()
# ConnectionStatus.CONNECTING, CONNECTED, DISCONNECTED, ERROR
# Force reconnect
provider.reconnect_sse()Polling Fallback
When SSE connection fails repeatedly, the SDK falls back to polling:
provider = FlipswitchProvider(
api_key="YOUR_API_KEY",
enable_polling_fallback=True, # default: True
polling_interval=30.0, # Poll every 30 seconds
max_sse_retries=5, # Fall back after 5 failed SSE attempts
)
# Check if polling is active
if provider.is_polling_active():
print("Polling fallback is active")Detailed Evaluation
Get full evaluation details including variant and reason:
details = client.get_boolean_details("feature-flag", False, context)
print(f"Value: {details.value}")
print(f"Variant: {details.variant}")
print(f"Reason: {details.reason}")Bulk Flag Evaluation
Evaluate all flags at once:
# Evaluate all flags
flags = provider.evaluate_all_flags(context)
for flag in flags:
print(f"{flag.key} ({flag.value_type}): {flag.get_value_as_string()}")
# Evaluate a single flag with full details
flag = provider.evaluate_flag("dark-mode", context)
if flag:
print(f"Value: {flag.value}")
print(f"Reason: {flag.reason}")
print(f"Variant: {flag.variant}")FastAPI Integration
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from flipswitch import FlipswitchProvider
from openfeature import api
from openfeature.evaluation_context import EvaluationContext
provider: FlipswitchProvider = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global provider
provider = FlipswitchProvider(api_key="YOUR_API_KEY")
api.set_provider(provider)
yield
provider.shutdown()
app = FastAPI(lifespan=lifespan)
@app.get("/feature")
def check_feature(user_id: str):
client = api.get_client()
context = EvaluationContext(targeting_key=user_id)
enabled = client.get_boolean_value("new-feature", False, context)
return {"feature_enabled": enabled}Django Integration
# settings.py
from flipswitch import FlipswitchProvider
from openfeature import api
FLIPSWITCH_PROVIDER = FlipswitchProvider(api_key="YOUR_API_KEY")
api.set_provider(FLIPSWITCH_PROVIDER)
# views.py
from openfeature import api
from openfeature.evaluation_context import EvaluationContext
def my_view(request):
client = api.get_client()
context = EvaluationContext(
targeting_key=str(request.user.id),
attributes={"email": request.user.email},
)
if client.get_boolean_value("new-feature", False, context):
# New feature logic
passReconnection Strategy
The SSE client automatically reconnects with exponential backoff:
- Initial delay: 1 second
- Maximum delay: 30 seconds
- Backoff multiplier: 2x
Shutdown
Always shutdown the provider when done:
provider.shutdown()In long-running applications (FastAPI, Django), make sure to call shutdown() when the application stops to close the SSE connection properly.
Troubleshooting
SSE connection keeps disconnecting
Check if your proxy or load balancer supports long-lived connections. Configure timeout settings accordingly.
Flags not updating in real-time
Verify that enable_realtime=True is set and check the SSE status:
print(provider.get_sse_status())Import errors
Make sure you have installed all required dependencies:
pip install flipswitch-sdk openfeature-sdk