Skip to content

Latest commit

 

History

History

README.md

cloudflare-flagship

PyPI version Python versions license

Flagship is a globally distributed, low-latency feature flag platform built entirely on Cloudflare. This package is the Python SDK — an OpenFeature-compliant provider for evaluating feature flags from Python server environments.

Note: The Python SDK supports HTTP mode only. The Cloudflare Workers binding mode (env.FLAGS) is exclusive to the TypeScript SDK and is not available in Python.

Installation

# uv
uv add cloudflare-flagship

# pip
pip install cloudflare-flagship

Quick start

from openfeature import api
from openfeature.evaluation_context import EvaluationContext
from flagship import FlagshipServerProvider

api.set_provider(
    FlagshipServerProvider(
        app_id="your-app-id",
        account_id="your-account-id",
        auth_token="your-token",
    )
)

client = api.get_client()
enabled = client.get_boolean_value(
    "dark-mode",
    False,
    EvaluationContext(targeting_key="user-123", attributes={"plan": "premium"}),
)

See examples/server.py for a full synchronous example and examples/async_server.py for async usage with asyncio.

Flag types

All four OpenFeature flag types are supported. Python's OpenFeature SDK splits the TypeScript number type into integer and float.

enabled = client.get_boolean_value("new-checkout", False, context)
variant = client.get_string_value("homepage-hero", "control", context)
limit   = client.get_integer_value("upload-limit", 10, context)
rate    = client.get_float_value("sample-rate", 0.1, context)
config  = client.get_object_value("ui-config", {"theme": "light"}, context)

Use the *_details variants when you need the full resolution result:

details = client.get_boolean_details("my-flag", False, context)

print(details.value)          # resolved value (or default on error)
print(details.reason)         # TARGETING_MATCH | SPLIT | DEFAULT | DISABLED | ERROR
print(details.variant)        # variation key, e.g. "on", "off", "v2"
print(details.error_code)     # set on error, e.g. FLAG_NOT_FOUND, TYPE_MISMATCH
print(details.error_message)

Configuration

FlagshipServerProvider accepts either app_id + account_id (recommended) or a full endpoint URL — not both.

FlagshipServerProvider(
    # Option A (recommended)
    app_id="your-app-id",
    account_id="your-account-id",

    # Option B: full URL (mutually exclusive with app_id)
    # endpoint="http://localhost:8787/v1/acct/apps/app-id/evaluate",

    # Static bearer token
    auth_token="your-token",

    # Dynamic credentials — called once per request, takes precedence over auth_token
    # headers_factory=lambda: {"Authorization": f"Bearer {get_token()}"},

    # Override the base URL for local dev
    # base_url="http://localhost:8787",

    timeout=5.0,      # seconds (default: 5.0)
    retries=1,        # retry attempts on transient errors, capped at 10 (default: 1)
    retry_delay=1.0,  # seconds between retries, capped at 30.0 (default: 1.0)
    logging=False,    # set True to enable SDK debug output (default: False)

    # Response caching — opt-in, off by default (see "Caching")
    # cache_ttl=30.0,      # seconds; enables caching when set
    # cache_max_size=1000, # max cached entries, LRU-evicted (default: 1000)
)
Option Type Default Description
app_id str Flagship app ID (mutually exclusive with endpoint)
account_id str Required with app_id
base_url str https://api.cloudflare.com Base URL override (only used with app_id)
endpoint str Full evaluation URL (mutually exclusive with app_id)
auth_token str Bearer token added to every request
headers_factory Callable[[], dict[str, str]] Called per request; takes precedence over auth_token
timeout float 5.0 Request timeout in seconds
retries int 1 Retry attempts on transient errors; capped at 10
retry_delay float 1.0 Delay between retries in seconds; capped at 30.0
logging bool False Enable SDK-level debug output via the flagship logger
cache_ttl float Cache TTL in seconds; enables caching when set
cache_max_size int 1000 Maximum cached entries; least-recently-used is evicted

Caching

The provider can cache evaluations to avoid a network round-trip for repeated flag/context pairs. Caching is off by default and enabled by setting cache_ttl (seconds):

FlagshipServerProvider(
    app_id="your-app-id",
    account_id="your-account-id",
    cache_ttl=30.0,        # cached values may be up to 30s stale
    cache_max_size=1000,   # LRU eviction beyond this many entries
)

Each entry is keyed by flag key, type, and the full evaluation context, so distinct contexts never share a value. Cache hits resolve with reason == Reason.CACHED. Disabled flags and errors are never cached. Because freshness is TTL-based, a flag change in Flagship takes effect after the entry expires.

The cache is shared by the sync and async APIs and guarded by a lock for thread-safe sync use.

Evaluation context

Context attributes are sent as URL query parameters. Supported types:

Type Serialisation
str, int, float Passed as a string
bool "true" or "false"
datetime ISO 8601
dict, list, other Not supported — raises InvalidContextError

Async

The async API mirrors the sync API — just await the *_async variants:

enabled = await client.get_boolean_value_async("dark-mode", False, context)
details = await client.get_boolean_details_async("dark-mode", False, context)

# Evaluate multiple flags concurrently
import asyncio
dark_mode, beta_access = await asyncio.gather(
    client.get_boolean_value_async("dark-mode", False, context),
    client.get_boolean_value_async("beta-access", False, context),
)

When shutting down in an async context, use shutdown_async() to properly close the HTTP client:

await api.shutdown_async()

Error handling

The provider never throws from a resolution method. On error the OpenFeature SDK returns the default value with an error_code and error_message.

Error code Cause
FLAG_NOT_FOUND Flag key does not exist (HTTP 404)
TYPE_MISMATCH The flag's resolved type does not match the requested type
INVALID_CONTEXT The evaluation context contains unsupported types (dict, list)
PARSE_ERROR The API response was not a valid evaluation response
GENERAL Network error, timeout, or any other transient failure

404 and 400 responses are never retried. All other failures are retried up to retries times.

Hooks

from flagship import LoggingHook, TelemetryHook

# Logs evaluation lifecycle events via the flagship logger (INFO level)
api.add_hooks([LoggingHook()])

# Emits a TelemetryEvent after every evaluation
api.add_hooks([TelemetryHook(lambda event: analytics.track("flag_evaluated", event))])

TelemetryEvent fields: type, flag_key, timestamp, duration_ms, value, reason, variant, error_code, error_message, context, hints.

Provider events

from openfeature.event import ProviderEvent

api.add_handler(ProviderEvent.PROVIDER_READY, lambda _: print("ready"))

Initialization does not perform network I/O. Flag evaluation requests happen only when resolving flags.

Development

uv sync --group dev        # install dependencies
uv run pytest              # run tests
uv run ty check            # type check
uv build                   # build wheel and sdist

License

Apache-2.0