Python SDK
Server-side feature flags for Python with built-in async support, circuit breaker, caching, and SSE streaming.
Circuit Breaker
Auto-recovery
Async Support
asyncio native
Caching
Persistent cache
SSE Support
Real-time updates
Installation
pip install rollgateQuick Start
from rollgate import RollgateClient
client = RollgateClient(
api_key="rg_server_...",
base_url="https://api.rollgate.io",
refresh_interval=30,
)
await client.init()
# Check flags
enabled = client.is_enabled("my-feature", False)
# Identify user
await client.identify({
"id": "user-123",
"email": "[email protected]",
"attributes": {"plan": "pro"},
})
# Cleanup
await client.close()Configuration
| Option | Type | Default | Description |
|---|---|---|---|
| api_key | str | required | Server API key (rg_server_...) |
| base_url | str | 'https://api.rollgate.io' | API base URL |
| refresh_interval | int | 30 | Polling interval in seconds, 0 to disable |
| enable_streaming | bool | False | Use SSE for real-time updates (opt-in) |
| timeout | int | 5 | Request timeout in seconds |
Methods
init()
Initialize the client. Fetches flags from the server and starts the polling loop. Must be awaited.
await client.init()close()
Shut down the client. Stops polling, closes SSE connections, and cleans up resources. Must be awaited.
await client.close()is_enabled(key, default_value)
Check if a boolean flag is enabled. Returns the flag value or the default.
enabled = client.is_enabled("new-feature", False)
if enabled:
# New feature code
passget_value(key, default_value)
Get a flag value of any type. Use for string, number, or JSON flags.
# String flag
theme = client.get_value("theme", "light")
# Number flag
max_retries = client.get_value("max-retries", 3)
# JSON flag
config = client.get_value("rate-limit-config", {"enabled": False, "limit": 100})get_string(key, default_value)
Type-safe helper for string flags.
button_color = client.get_string("cta-color", "blue")
variant = client.get_string("checkout-variant", "control")get_number(key, default_value)
Type-safe helper for number flags.
page_size = client.get_number("results-per-page", 20)
timeout = client.get_number("api-timeout-ms", 5000)get_json(key, default_value)
Type-safe helper for JSON flags. Returns a dictionary.
config = client.get_json("feature-config", {
"enabled": False,
"max_items": 10,
"allowed_roles": ["admin"],
})
# Access properties
if config["enabled"] and user.role in config["allowed_roles"]:
# Feature logic
passget_all_flags()
Get all flags as a dictionary.
flags = client.get_all_flags()
# {"feature-a": True, "feature-b": False}identify(user)
Set user context for targeting rules. Re-fetches flags with the new context. Must be awaited.
await client.identify({
"id": "user-123",
"email": "[email protected]",
"attributes": {
"plan": "enterprise", # For "plan in pro,enterprise" rules
"country": "US", # For "country equals US" rules
"company": "Acme Inc", # For "company contains Acme" rules
"app_version": "2.1.0", # For "app_version semver_gt 2.0.0" rules
},
})refresh()
Force refresh flags from the server. Must be awaited.
await client.refresh()Async Usage
The Python SDK is built on asyncio. All network operations (init, close, identify, refresh) are async and must be awaited. Flag evaluation methods (is_enabled, get_value, etc.) are synchronous and read from the local cache.
import asyncio
from rollgate import RollgateClient
async def main():
client = RollgateClient(
api_key="rg_server_...",
base_url="https://api.rollgate.io",
)
await client.init()
# Synchronous flag checks (reads from cache)
enabled = client.is_enabled("my-feature", False)
theme = client.get_string("theme", "dark")
print(f"Feature enabled: {enabled}")
print(f"Theme: {theme}")
await client.close()
asyncio.run(main())Circuit Breaker
The SDK includes a circuit breaker that protects your app when Rollgate is unavailable. When open, it uses cached flag values.
# Get current state
state = client.get_circuit_state()
# "closed" | "open" | "half-open"
# Listen to events
@client.on("circuit-open")
def on_circuit_open():
print("Circuit breaker opened - using cached flags")
@client.on("circuit-closed")
def on_circuit_closed():
print("Circuit breaker closed - normal operation")
# Force reset (use with caution)
client.reset_circuit()Framework Examples
FastAPI
from fastapi import FastAPI
from rollgate import RollgateClient
app = FastAPI()
rollgate = RollgateClient(api_key="rg_server_...")
@app.on_event("startup")
async def startup():
await rollgate.init()
@app.on_event("shutdown")
async def shutdown():
await rollgate.close()
@app.get("/api/feature")
async def feature():
enabled = rollgate.is_enabled("new-feature", False)
return {"enabled": enabled}Flask
import asyncio
from flask import Flask, jsonify
from rollgate import RollgateClient
app = Flask(__name__)
rollgate = RollgateClient(api_key="rg_server_...")
# Initialize on startup
loop = asyncio.new_event_loop()
loop.run_until_complete(rollgate.init())
@app.route("/api/feature")
def feature():
enabled = rollgate.is_enabled("new-feature", False)
return jsonify({"enabled": enabled})
@app.teardown_appcontext
def shutdown(exception=None):
loop.run_until_complete(rollgate.close())Django
import asyncio
from rollgate import RollgateClient
# Initialize as a singleton
_loop = asyncio.new_event_loop()
rollgate = RollgateClient(api_key="rg_server_...")
_loop.run_until_complete(rollgate.init())from django.http import JsonResponse
from .flags import rollgate
def feature_view(request):
enabled = rollgate.is_enabled("new-feature", False)
return JsonResponse({"enabled": enabled})Error Handling & Graceful Degradation
The SDK is designed to never raise exceptions during flag evaluation. If Rollgate is unavailable, flags return their default values.
# Default values ensure your app works even without Rollgate
show_new_feature = client.is_enabled("new-feature", False)
# Returns False if: flag doesn't exist, API down, circuit open
# Monitor degraded state
@app.get("/health")
async def health():
circuit = client.get_circuit_state()
is_healthy = circuit == "closed"
return {
"status": "healthy" if is_healthy else "degraded",
"circuit": circuit,
"flags_source": "cache" if circuit == "open" else "live",
}