Python SDK Reference

TL;DR: Install featureflip, call client.variation("flag-key", {"user_id": "u-123"}, default=False), and you’re evaluating flags — synchronous, no asyncio.

The featureflip Python package provides a synchronous client for evaluating feature flags in any Python 3.10+ application — Django, Flask, FastAPI, Celery workers, scripts, or notebooks. The SDK is intentionally synchronous: variation calls do not return coroutines, so you can drop them into existing sync codebases without rewriting call sites or introducing asyncio to a non-async project.

Internally the client uses httpx for HTTP and SSE streaming, and stores flag configurations in a thread-safe in-memory cache. Evaluations read from this cache under a lock and return immediately, so they are safe to call from gunicorn workers, threaded WSGI servers, and Celery prefork pools alike. There is no GIL contention because the hot path is a dictionary lookup, not Python bytecode crunching JSON. Under a preforking server, build the client in the worker rather than the master (see Lifetime).

Evaluation context is a plain dict[str, Any] rather than a class — {"user_id": "u-123", "plan": "pro"} is the entire API. The user_id key is reserved: it anchors percentage rollouts so a given user lands in a stable bucket across processes and pods. Custom keys feed into targeting rules. String enum values from the API (rule operators, evaluation reasons) are normalized to lowercase on the client side, so you can compare them against Python literals without worrying about casing.

New to Featureflip?

Start with the Python quickstart — it walks through installation, initializing the client, and evaluating your first flag in under 5 minutes.

View source on GitHub

Installation

pip install featureflip

The package depends on httpx, httpx-sse, and structlog.

Lifetime

The Featureflip Python SDK is designed to be used as a singleton across your application. Construct the client once and share it:

# In your application startup
from featureflip import FeatureflipClient

client = FeatureflipClient(sdk_key="your-sdk-key")

Calling FeatureflipClient(sdk_key="x") multiple times with the same SDK key returns distinct handle objects that share a single underlying client — the SDK maintains a process-wide cache keyed by SDK key. You cannot accidentally open duplicate streaming connections by constructing multiple clients.

# Both of these return distinct handle objects that share one underlying client
c1 = FeatureflipClient(sdk_key="your-sdk-key")
c2 = FeatureflipClient(sdk_key="your-sdk-key")
assert c1 is not c2           # different handles
# ...but only one HTTP connection, one polling thread, one event processor

c1.close()  # decrements refcount; c2 is still alive
c2.close()  # last handle closed; real shutdown runs

Different SDK keys get independent clients:

client_a = FeatureflipClient(sdk_key="sdk-key-a")
client_b = FeatureflipClient(sdk_key="sdk-key-b")
# Two independent clients

Use with Django, Flask, FastAPI

Build the client inside each worker process, after the fork. The constructor starts background threads for the SSE stream and the analytics flush, and os.fork() copies only the calling thread. A client built in a preloading master hands every worker a flag store that no thread is updating, so evaluations keep returning whatever was true at fork time and nothing raises.

The portable pattern defers construction to first use, which lands it in the worker under every server:

myapp/flags.py

import os
from featureflip import FeatureflipClient

_client: FeatureflipClient | None = None

def get_client() -> FeatureflipClient:
    global _client
    if _client is None:
        _client = FeatureflipClient(sdk_key=os.environ["FEATUREFLIP_SDK_KEY"])
    return _client

Django’s AppConfig.ready() and a Flask create_app() factory both run per worker under a plain gunicorn myproject.wsgi, so calling get_client() from either is fine. Add --preload and the same code runs once in the master instead, and uWSGI preloads by default unless you set lazy-apps = true. Where you want construction pinned to the worker regardless of launch flags, use the server’s post-fork hook: Gunicorn’s post_fork in gunicorn.conf.py, uWSGI’s @postfork decorator, or a FastAPI lifespan context manager, which already runs per worker.

The cache solves a different problem. Two FeatureflipClient(...) calls in one process share one connection, so duplicate construction is harmless. It cannot restart a thread the fork left behind.

Feature flags in Python works through the failure mode and each server’s hook in detail.

Closing

close() is refcounted. When multiple handles share one cached client, closing one handle does not shut down the shared background threads — the real shutdown runs only when the last handle is closed. In a typical application the client lives for the entire process lifetime and is cleaned up automatically at process exit or via a framework’s shutdown lifecycle.

As a context manager:

with FeatureflipClient(sdk_key="your-sdk-key") as client:
    value = client.variation("flag", {"user_id": "123"}, default=False)
# client.close() is called automatically; decrements refcount

Configuration

The Config dataclass accepts the following options:

Option	Type	Default	Description
base_url	str	"https://eval.featureflip.io"	Base URL of the Evaluation API
connect_timeout	float	5.0	Connection timeout in seconds
read_timeout	float	10.0	Read timeout in seconds
streaming	bool	True	Enable SSE for real-time flag updates
poll_interval	float	30.0	Polling interval in seconds (used when streaming is disabled)
send_events	bool	True	Enable analytics event tracking
flush_interval	float	30.0	Interval in seconds between event flush batches
flush_batch_size	int	100	Maximum number of events per flush batch
init_timeout	float	10.0	Maximum time in seconds to wait for initialization. With snapshot_store, a failed fetch is followed by a store read capped at the same time, so the worst case is twice this value
snapshot_store	SnapshotStore | None	None	Persists the last flag configuration so a process that starts while the Evaluation API is unreachable can still serve real values. See Starting during an outage
max_snapshot_age_seconds	float	86400	Oldest snapshot, in seconds, the client will start from. Must be a positive, finite number. Only used with snapshot_store
wrapper_name	str | None	None	For libraries that wrap the SDK: a name appended to the User-Agent after the SDK’s own token
wrapper_version	str | None	None	Version sent with wrapper_name

Example configuration

from featureflip import Config

config = Config(
    base_url="https://eval.featureflip.io",
    streaming=True,
    poll_interval=60.0,
    flush_interval=15.0,
    init_timeout=30.0,
)

All timeout and interval values must be positive. The constructor validates these constraints and raises ValueError for invalid values.

Initialization

Create a client by providing an SDK key and optional configuration:

from featureflip import FeatureflipClient, Config

client = FeatureflipClient(
    sdk_key="sdk-dev-abc123",
    config=Config(base_url="https://eval.featureflip.io"),
)

The constructor performs initialization synchronously: it fetches the initial flag configuration, starts streaming or polling, and begins the event processor. If initialization fails or times out, an InitializationError is raised. With a snapshot_store set, a failed start does not raise. The client starts from the stored snapshot, or serves your defaults when there is none, and keeps retrying in the background (see Starting during an outage).

SDK key resolution

If sdk_key is not provided, the client falls back to the FEATUREFLIP_SDK_KEY environment variable. A ValueError is raised if neither is available.

import os
os.environ["FEATUREFLIP_SDK_KEY"] = "sdk-dev-abc123"

# SDK key read from environment
client = FeatureflipClient()

Context manager

The client supports Python’s context manager protocol for automatic cleanup:

with FeatureflipClient(sdk_key="sdk-dev-abc123") as client:
    enabled = client.variation("new-feature", {"user_id": "123"}, default=False)
# client.close() is called automatically

is_initialized

client.is_initialized: bool

A read-only property that returns True once the client has successfully loaded flag data, or has finished starting with a snapshot_store configured. Check startup_info to tell a live start from a snapshot start.

Starting during an outage

By default the constructor raises InitializationError when the Evaluation API is unreachable at startup. Set snapshot_store to let a new process start from the last configuration it saved instead. The option is off unless you set it. For how this fits with the rest of the failure behavior, see Reliability.

from featureflip import Config, FeatureflipClient, FileSnapshotStore

client = FeatureflipClient(
    sdk_key="sdk-dev-abc123",
    config=Config(
        snapshot_store=FileSnapshotStore("/var/lib/featureflip/snapshot.json"),
        max_snapshot_age_seconds=6 * 60 * 60,
    ),
)

How it behaves:

• The snapshot is read only when the initial fetch fails or times out. A successful fetch always wins, and the snapshot is never consulted afterward.
• When the fetch fails and no usable snapshot exists, the constructor returns a client that serves your defaults instead of raising. startup_info says why. If you relied on except InitializationError to fail fast, check client.startup_info.source instead, because with a snapshot store configured the constructor no longer raises.
• After a start from a snapshot the client keeps retrying the live API and replaces the snapshot’s values as soon as it connects. Listeners already registered at that moment fire for what changed between the two. The constructor blocks, so a listener you add right after it returns can miss that first change. Read startup_info after registering if you need to know.
• The client writes the full configuration after each full update it applies (the first fetch, a poll, a stream resync). Once it is connected live, single-flag updates write it too. Writes run on a background thread and never block evaluation.
• While the client stays connected and healthy, it refreshes the stored copy at most every 5 minutes even when nothing changed, so a quiet configuration does not age past max_snapshot_age_seconds. Changes are written right away.
• The store read is bounded by init_timeout. If the fetch times out and then the store read hangs too, the constructor returns after twice init_timeout, serving defaults.
• The option is read on the first client built for an SDK key. Later clients with the same key reuse the existing one, as described under Lifetime, and log a warning if they pass a different snapshot_store or set a different max_snapshot_age_seconds. Leaving the age out never warns.

A snapshot contains your targeting rules, including any customer attribute values used in them. On POSIX systems FileSnapshotStore creates the file readable by its owner only (mode 0600). It does not call fsync before replacing the file, so a power loss can leave an empty file. The client reads that as snapshot-invalid, logs a warning and serves defaults until the next write.

Container disks are usually ephemeral, so a file written inside the image or container layer is gone after a redeploy. Point the path at a mounted volume:

docker-compose.yml

services:
  api:
    volumes:
      - featureflip-snapshot:/var/lib/featureflip
volumes:
  featureflip-snapshot:

On Kubernetes, mount a PersistentVolumeClaim, or an emptyDir if you only need the file to survive container restarts within a pod, at the same path.

FileSnapshotStore won’t create the parent directory, so it must already exist. Check the owner too. Docker creates a new named volume owned by root unless the image already has that directory, so a process running as a non-root user fails every write and logs a warning each time. Create the directory in your Dockerfile and hand it to that user, for example RUN mkdir -p /var/lib/featureflip && chown app /var/lib/featureflip.

startup_info

client.startup_info reports where the flag data came from. It is a frozen StartupInfo dataclass:

Field	Values	Meaning
source	"live", "snapshot", "none"	live: the initial fetch succeeded, or the client has since connected after starting from snapshot or none. snapshot: the client started from the store. none: no data yet, so evaluations return your defaults
reason	"snapshot", "snapshot-expired", "snapshot-invalid", or None	Set when a snapshot was considered, and cleared once the client is live. snapshot: it was used. snapshot-expired: it was older than max_snapshot_age_seconds. snapshot-invalid: the stored text could not be used. That covers a garbage, empty or cut-off file, a snapshot written for a different SDK key or base URL, and a timestamp more than 60 seconds in the future. An empty store sets no reason
snapshot_age_seconds	float | None	Age of the snapshot that was used or rejected for age

info = client.startup_info
if info.source != "live":
    logger.warning(
        "flags did not load from the API",
        extra={"source": info.source, "reason": info.reason, "age": info.snapshot_age_seconds},
    )

Custom stores

To share one snapshot across processes, implement SnapshotStore. It stores text: get() returns the string last stored, or None when there is none, and put(text) replaces it. Both are synchronous and may be called from a background thread. This sketch uses Redis:

import os

import redis

from featureflip import SnapshotStore

r = redis.Redis.from_url(os.environ["REDIS_URL"], decode_responses=True)
KEY = "featureflip:snapshot:production"


class RedisSnapshotStore:
    def get(self) -> str | None:
        return r.get(KEY)

    def put(self, text: str) -> None:
        r.set(KEY, text)


store: SnapshotStore = RedisSnapshotStore()

The store never needs to parse the text. The SDK wraps the configuration in its own envelope, with a timestamp and a hash of the SDK key and base URL, and checks all of it on read. Text written for a different environment is rejected as snapshot-invalid and never served, which is why one store instance can safely be shared by clients for several SDK keys. Still give each environment its own key or path. Two environments sharing one slot overwrite each other’s snapshot.

A store that raises is treated as having no snapshot, or as a failed write, and the client logs a warning.

Evaluating Flags

The Python SDK uses a single variation() method for all flag types, rather than separate typed methods.

variation()

enabled = client.variation("dark-mode", {"user_id": "u-123"}, default=False)
plan = client.variation("pricing-tier", {"user_id": "u-123"}, default="free")
limit = client.variation("rate-limit", {"user_id": "u-123"}, default=100)

Signature:

def variation(
    self,
    key: str,
    context: dict[str, Any],
    default: T,
    track: bool = True,
) -> T

Parameter	Type	Description
key	str	The flag key to evaluate
context	dict[str, Any]	User attributes for targeting. The user_id key is used for percentage rollouts
default	T	Value returned if the flag is not found or evaluation fails
track	bool	Whether to track this evaluation event (default True)

This method never raises exceptions. On any error, it logs a warning and returns the default value.

Evaluation Details

variation_detail()

Returns an EvaluationDetail object with the evaluated value and metadata about the evaluation decision.

detail = client.variation_detail("dark-mode", {"user_id": "u-123"}, default=False)

print(detail.value)    # True
print(detail.reason)   # EvaluationReason.RULE_MATCH
print(detail.rule_id)  # "rule-abc"

Signature:

def variation_detail(
    self,
    key: str,
    context: dict[str, Any],
    default: T,
) -> EvaluationDetail

EvaluationDetail

A frozen dataclass containing the evaluation result:

Property	Type	Description
value	Any	The evaluated flag value
reason	EvaluationReason	Why this value was returned
rule_id	str | None	The ID of the matched targeting rule, if applicable
error	Exception | None	The exception that occurred, if reason is ERROR

EvaluationReason

An enum describing why a particular flag value was returned:

Value	Description
FALLTHROUGH	No rules matched; the fallthrough variation was served
RULE_MATCH	A targeting rule matched the context
FLAG_DISABLED	The flag is disabled; the off variation was served
FLAG_NOT_FOUND	The flag key does not exist
ERROR	An error occurred during evaluation

Event Tracking

track()

Records a custom analytics event.

client.track("purchase-completed", {"user_id": "u-123"}, metadata={
    "plan": "pro",
    "amount": 29.99,
})

Signature:

def track(
    self,
    event_name: str,
    context: dict[str, Any],
    metadata: dict[str, Any] | None = None,
) -> None

Featureflip counts each event name, per environment. The metadata you pass is transmitted with the event but is not stored, and event counts are not surfaced in the app or API.

identify()

Records an Identify event for the given user. The event is counted per environment, and the attributes you pass are not stored.

This does not affect flag evaluation. Targeting rules and segments are matched against the context you pass to each variation call. identify() does not set a context for subsequent calls, and it does not persist user attributes for later targeting.

client.identify({
    "user_id": "u-123",
    "email": "user@example.com",
    "plan": "pro",
})

Signature:

def identify(self, context: dict[str, Any]) -> None

Flag Change Notifications

on_update()

Subscribes to flag-configuration updates. The listener receives the keys of the flags affected by each change, batched into one call however many flags it touched.

def on_flags_changed(keys: list[str]) -> None:
    print("flags changed:", keys)
    cache.invalidate(keys)

unsubscribe = client.on_update(on_flags_changed)

A flag is reported when it is created, deleted or redefined, when a segment its targeting rules reference changes, and when a flag it depends on through a prerequisite changes. Those last two move a flag’s evaluated value without touching its own configuration.

The listener does not fire for the initial flag load, and is never called with an empty list.

Listeners run on the SDK’s streaming or polling thread, so they must not block: a slow listener delays flag delivery and, on the streaming path, can stall the connection. Do the minimum here and hand real work to a queue or thread pool.

Exceptions raised by a listener are caught and logged. A throwing listener neither breaks flag delivery nor stops the remaining listeners.

Returns a callable that unsubscribes. Listeners registered through a handle are also dropped when that handle is closed, so a handle that goes away cannot leave a listener firing off the shared core.

Signature:

on_update(listener: Callable[[list[str]], None]) -> Callable[[], None]

off_update()

Removes a listener previously registered with on_update(). Equivalent to calling the unsubscribe callable it returned. Removing a listener that is not registered is a no-op.

client.off_update(on_flags_changed)

Cleanup

close()

Shuts down the client, stops streaming or polling, flushes remaining events, and closes HTTP connections.

client.close()

flush()

Forces all buffered events to be sent to the server immediately.

client.flush()

Using the context manager (with statement) is the recommended approach, as it calls close() automatically when the block exits.

Testing

FeatureflipClient.for_testing()

Creates a test client with fixed flag values. No network calls are made and no background threads are started.

client = FeatureflipClient.for_testing({
    "dark-mode": True,
    "pricing-tier": "enterprise",
    "rate-limit": 500,
})

client.variation("dark-mode", {}, default=False)      # True
client.variation("pricing-tier", {}, default="free")   # "enterprise"

Signature:

@classmethod
def for_testing(cls, flags: dict[str, Any]) -> FeatureflipClient

set_test_value()

Updates a flag value on a test client at runtime. Raises RuntimeError if called on a non-test client.

client = FeatureflipClient.for_testing({"dark-mode": True})

# Change the value during the test
client.set_test_value("dark-mode", False)
client.variation("dark-mode", {}, default=True)  # False

Signature:

def set_test_value(self, key: str, value: Any) -> None

Types

Public exports

All public types are available from the top-level featureflip package:

from featureflip import (
    FeatureflipClient,
    Config,
    EvaluationContext,
    EvaluationDetail,
    EvaluationReason,
    FlagUpdateListener,
    FileSnapshotStore,
    SnapshotStore,
    StartupInfo,
    InitializationError,
    ConfigurationError,
    FeatureflipError,
)

Exception hierarchy

Exception	Description
FeatureflipError	Base exception for all SDK errors
InitializationError	Raised when initialization fails or times out
ConfigurationError	Raised for invalid configuration values

See also

• Python Quickstart — get started in under 5 minutes
• Targeting & Segments — control which users see which variation
• Rollout Strategies — gradual percentage-based rollouts
• Evaluation API — REST API reference
• OpenFeature — use Featureflip through the vendor-neutral OpenFeature standard
• SDK Overview — compare server-side and client-side SDKs