Unleash SDK for Python

Unleash is a private, secure, and scalable feature management platform built to reduce the risk of releasing new features and accelerate software development. This server-side Python SDK is designed to help you integrate with Unleash and evaluate feature flags inside your application.

You can use this client with Unleash Enterprise or Unleash Open Source.

Migrating to v7

The client's settings are now grouped into option objects, and every argument after url and app_name is keyword-only. Read the v7 migration guide before upgrading.

Migrating to v6

If you use custom strategies or access the features property on the Unleash Client, read the complete migration guide before upgrading to v6.

Getting Started

Install the Unleash Client in your project

pip install UnleashClient

Initialization

You must initialize the SDK before you use it. Note that until the SDK has synchronized with the API, all features will evaluate to false unless you have a bootstrapped configuration or you use fallbacks.

from UnleashClient import UnleashClient

client = UnleashClient(
    url="https:<YOUR-API-URL>",
    app_name="my-python-app",
    api_key="<API token>")

client.initialize_client()

Check features

Once the SDK is initialized, you can evaluate toggles using the is_enabled or get_variant methods.

enabled = client.is_enabled("my_toggle")
print(enabled)
> True

variant = client.get_variant("variant_toggle")
print(variant)
> {
>    "name": "variant1",
>    "payload": {
>        "type": "string",
>        "value": "val1"
>        },
>    "enabled": True
> }

Shutdown

If your program no longer needs the SDK, you can call destroy(), which shuts down the SDK and flushes any pending metrics to Unleash.

client.destroy()

Usage

Context

Both the is_enabled and get_variant functions support Unleash contexts as the second parameter.

app_context = {
    "userId": "[email protected]",
    "sessionId": "55845",
    "properties": {
        "custom-property": "some-value"
    }
}

client.is_enabled("user_id_toggle", app_context)
client.get_variant("variant_toggle", app_context)

The context values can be any type that has a __str__ implementation. Types that are explicitly supported are:

Numerics
Strings
Dates
UUIDs

Note: Gradual rollout strategies require you to pass either a userId or a sessionId for stickiness to work correctly.

Fallback function

You can specify a fallback function for cases where the client doesn't recognize the toggle by using the fallback_function keyword argument:

def custom_fallback(feature_name: str, context: dict) -> bool:
    return True

client.is_enabled("my_toggle", fallback_function=custom_fallback)

The fallback function must accept the feature name and context as positional arguments in that order.

The client will evaluate the fallback function if the feature flag is not found or an exception occurs when calling the is_enabled() method.

Configuration options

url and app_name are required. Everything else is a keyword argument, and related settings are grouped into option objects:

from UnleashClient import (
    MetricsOptions, PollingOptions, RequestsTransport, UnleashClient,
)

client = UnleashClient(
    url="https://YOUR-API-URL",
    app_name="my-python-app",
    api_key="<API token>",
    transport=RequestsTransport(timeout=10),
    mode=PollingOptions(interval=30),
    metrics=MetricsOptions(disabled=True),
)

Settings can't be changed after the client is created.

Parameter	Description	Default
url	URL of your Unleash server, e.g. `https://app.unleash-hosted.com/demo/api/`. Required.	—
app_name	Name of the application using the client. Required.	—
api_key	Client API token, sent as the `Authorization` header.	None
identity	How the client identifies itself. See `IdentityOptions`.	`IdentityOptions()`
mode	How the client receives flag updates: `PollingOptions` or `StreamingOptions`.	`PollingOptions()`
metrics	How the client reports to the server. See `MetricsOptions`.	`MetricsOptions()`
verbose_log_level	Log level for messages about unknown flags and failed impression events.	30
custom_strategies	Dict of `{name: strategy}` for custom activation strategies. See custom strategies.	None
cache	Custom cache implementation (must extend `BaseCache`). See custom cache.	`FileCache`
cache_directory	Directory for the default `FileCache`. When unset, a platform-specific user cache directory is used. Ignored when `cache` is given.	None
scheduler	Custom APScheduler instance. When unset, the client creates a `BackgroundScheduler`. Sync client only.	None
scheduler_executor	Name of the executor to use on your custom `scheduler`. Required with `scheduler`. Sync client only.	None
multiple_instance_mode	How to handle multiple client instances (`BLOCK`, `WARN`, `SILENTLY_ALLOW`).	`WARN`
event_callback	Function that receives impression and lifecycle events. See events and impression data.	None
transport	How requests are sent, including headers, timeout, proxy and TLS settings. See custom HTTP transport.	`RequestsTransport` or `AiohttpTransport`

`IdentityOptions`

Parameter	Description	Default
instance_id	Unique identifier for this client instance.	"unleash-python-sdk"
sdk_flavor	Identifier of an integration built on this SDK (e.g. an OpenFeature provider).	None
sdk_flavor_version	Version of the integration named by `sdk_flavor`.	None

`PollingOptions`

The default: the client fetches feature flags from the server on an interval.

Parameter	Description	Default
interval	How often to fetch feature flags (seconds).	15
jitter	Maximum random delay added to each refresh (seconds).	None
project_name	Only fetch the feature flags of this project.	None

`StreamingOptions`

Experimental: the client keeps a streaming connection open and receives flag updates as they happen. Pass mode=StreamingOptions(). It has no settings yet. AsyncUnleashClient doesn't support streaming.

`MetricsOptions`

Parameter	Description	Default
interval	How often to send metrics to Unleash (seconds).	60
jitter	Maximum random delay added to each metrics send (seconds).	None
disabled	Disable sending usage metrics.	False
disable_registration	Disable client registration.	False

Bootstrap

By default, the Python SDK fetches your feature flags from the Unleash API at startup. If you want to make your SDK more resilient (e.g., during network outages), you can bootstrap the client with a local or remote toggle config.

How it works:

Use a FileCache (or your own BaseCache implementation).
Pre-seed it with feature flags using bootstrap_from_dict, bootstrap_from_file, or bootstrap_from_url.
Pass your cache to the UnleashClient on startup.

The default FileCache has built-in methods for bootstrapping from a dictionary, file, or URL.

Bootstrap from dict

from UnleashClient.cache import FileCache
from UnleashClient import UnleashClient

# Create and seed the cache
cache = FileCache("MY_CACHE")
cache.bootstrap_from_dict({
    "version": 2,
    "features": [
        {
            "name": "my_toggle",
            "enabled": True,
            "strategies": [{"name": "default"}],
        }
    ]
})

client = UnleashClient(
    url="https://YOUR-API-URL",
    app_name="my-python-app",
    cache=cache
)

Bootstrap from file

from pathlib import Path
from UnleashClient.cache import FileCache

cache = FileCache("MY_CACHE")
cache.bootstrap_from_file(Path("/path/to/your_bootstrap.json"))

client = UnleashClient(
    url="https://YOUR-API-URL",
    app_name="my-python-app",
    cache=cache
)

Bootstrap from URL

from UnleashClient.cache import FileCache

cache = FileCache("MY_CACHE")
cache.bootstrap_from_url("https://your-server/bootstrap.json")

client = UnleashClient(
    url="https://YOUR-API-URL",
    app_name="my-python-app",
    cache=cache
)

Custom strategies

The Python SDK lets you define custom activation strategies if the built-in ones don’t cover your needs. This gives you more fine grained control over how your features evaluate.

A custom strategy is just a class that implements an apply method.

class ActiveForEmailStrategy:
    def apply(self, parameters: dict, context: dict = None) -> bool:
        # Decide if the feature is active for this context
        return context.get("email") in parameters

Once you’ve defined your strategy, register it when you initialize the client. The key must match the strategy name in Unleash exactly.

## You should have a custom strategy defined in Unleash called 'EmailStrategy'
my_custom_strategies = {
    "EmailStrategy": ActiveForEmailStrategy()
}

client = UnleashClient(

    url="https://YOUR-API-URL",
    app_name="my-python-app",
    api_key="<API token>",
    custom_strategies = my_custom_strategies
)

Events and impression data

The Python SDK lets you tap into its behavior through impression data and lifecycle events.

Note: The SDK does not include a built-in event bus — you’ll need to provide your own. The example below shows how to use Blinker to send signals.

Impression events

To use impression data:

Enable impression data on your feature flags in the Unleash UI.
Provide an event_callback function when you initialize the client.

Your callback must accept a single UnleashEvent. You can log it, store it, or send it to another system.

from blinker import signal
from UnleashClient import UnleashClient
from UnleashClient.events import UnleashEvent

send_data = signal('send-data')

@send_data.connect
def receive_data(sender, **kw):
    print("Caught signal from %r, data %r" % (sender, kw))

def example_callback(event: UnleashEvent):
    send_data.send('anonymous', data=event)

client = UnleashClient(
    url="https://YOUR-API-URL",
    app_name="my-python-app",
    api_key="<API token>",
    event_callback=example_callback
)
client.initialize_client()
client.is_enabled("testFlag")

Impression callbacks run in-process — keep them fast to avoid blocking your app.

Lifecycle events

The same event_callback also delivers lifecycle events:

FETCHED: triggered when a new version of feature flags is pulled from the Unleash server. (Does not trigger on 304 Not Modified). The FETCHED event includes a features property containing all the feature flags returned by that fetch.
READY: triggered once when the SDK first loads feature flags from the Unleash server or a local backup.

from blinker import signal
from UnleashClient import UnleashClient
from UnleashClient.events import UnleashEvent, UnleashEventType

send_data = signal('send-data')

@send_data.connect
def receive_data(sender, **kw):
    if kw["data"].event_type == UnleashEventType.READY:
        print("SDK is ready: toggles loaded from Unleash or backup")
    elif kw["data"].event_type == UnleashEventType.FETCHED:
        # Only FETCHED events have a 'features' property
        print("Fetched new feature flags:", kw["data"].features)

def example_callback(event: UnleashEvent):
    send_data.send('anonymous', data=event)

client = UnleashClient(
    url="https://YOUR-API-URL",
    app_name="my-python-app",
    api_key="<API token>",
    event_callback=example_callback
)
client.initialize_client()
client.is_enabled("testFlag")

Impact metrics

Impact metrics are lightweight, application-level time-series metrics stored and visualized directly inside Unleash. They allow you to connect specific application data, such as request counts, error rates, or latency, to your feature flags and release plans.

These metrics help validate feature impact and automate release processes. For instance, you can monitor usage patterns or performance to determine if a feature meets its goals.

The SDK automatically attaches context labels to metrics: appName and environment. The environment is taken from your API token, or is "default" when the token doesn't name one.

Counters

Use counters for cumulative values that only increase (total requests, errors):

client.impact_metrics.define_counter(
    "request_count",
    "Total number of HTTP requests processed"
)

client.impact_metrics.increment_counter("request_count")

Gauges

Use gauges for point-in-time values that can go up or down:

client.impact_metrics.define_gauge(
    "total_users",
    "Total number of registered users"
)

client.impact_metrics.update_gauge("total_users", user_count)

Histograms

Histograms measure value distribution (request duration, response size):

client.impact_metrics.define_histogram(
    "request_time_ms",
    "Time taken to process a request in milliseconds",
    [50, 100, 200, 500, 1000]
)

client.impact_metrics.observe_histogram("request_time_ms", 125)

Impact metrics are batched and sent using the same interval as standard SDK metrics.

Custom cache

By default, the Python SDK stores feature flags in an on-disk cache using fcache. If you need a different storage backend, for example, Redis, memory-only, or a custom database, you can provide your own cache implementation.

Below is an example custom CustomCache using fcache under the hood.

from typing import Optional, Any
from UnleashClient.cache import BaseCache
from fcache.cache import FileCache as _FileCache

class CustomCache(BaseCache):
    # This is specific for FileCache.  Depending on the cache you're using, this may look different!
    def __init__(self, name: str, directory: Optional[str] = None):
        self._cache = _FileCache(name, app_cache_dir=directory)

    def set(self, key: str, value: Any):
        self._cache[key] = value
        self._cache.sync()

    def mset(self, data: dict):
        self._cache.update(data)
        self._cache.sync()

    def get(self, key: str, default: Optional[Any] = None):
        return self._cache.get(key, default)

    def exists(self, key: str):
        return key in self._cache

    def destroy(self):
        return self._cache.delete()

Pass your cache instance to the client with the cache argument:

client = UnleashClient(
    url="https://YOUR-API-URL",
    app_name="my-python-app",
    cache=CustomCache("my-cache")
)

Custom HTTP transport

By default, UnleashClient sends its requests with requests, on a new session for every request, and AsyncUnleashClient with one aiohttp session. Pass transport= to choose how requests are sent:

RequestsTransport(session) or AiohttpTransport(session) to keep the default library but configure the session;
any object with an execute(request) -> Response method, to use another HTTP library. For AsyncUnleashClient, execute is async.

Custom `requests` session

The session is used for every request and never closed by the client: close it yourself after client.destroy().

import requests
from UnleashClient import RequestsTransport, UnleashClient

session = requests.Session()

# examples
# proxy
session.proxies = {"https": "http://proxy.internal:3128"}

# A private CA, or a client certificate for mutual TLS
session.verify = "/etc/ssl/certs/internal-ca.pem"
session.cert = ("/etc/unleash/client.crt", "/etc/unleash/client.key")

# hooks
session.hooks["response"].append(
    lambda resp, *args, **kwargs: metrics.observe(resp.elapsed.total_seconds())
)

client = UnleashClient(
    url="https://YOUR-API-URL",
    app_name="my-python-app",
    api_key="<API token>",
    transport=RequestsTransport(
        session,
        headers={"X-Team": "checkout"},  # sent with every request
        timeout=10,                      # seconds, for the connection and each read
    ),
)

A session also keeps connections open between polls, instead of opening a new one each time.

RequestsTransport takes headers and timeout with or without a session. The SDK's own headers take precedence over custom ones, and an Authorization header raises ValueError: pass the API token as api_key.

Custom `aiohttp` session

The same for AsyncUnleashClient, with an aiohttp.ClientSession. Create it inside a coroutine, since aiohttp needs a running event loop, and close it after the client:

import ssl
import aiohttp
from UnleashClient import AiohttpTransport
from UnleashClient.clients.async_unleash_client import AsyncUnleashClient

# examples
tls = ssl.create_default_context(cafile="/etc/ssl/certs/internal-ca.pem")
tls.load_cert_chain("/etc/unleash/client.crt", "/etc/unleash/client.key")

async def main():
    async with aiohttp.ClientSession(
        trust_env=True,
        connector=aiohttp.TCPConnector(ssl=tls),
        trace_configs=[my_trace_config],  # e.g. aiohttp tracing for OpenTelemetry
    ) as session:
        async with AsyncUnleashClient(
            url="https://YOUR-API-URL",
            app_name="my-python-app",
            api_key="<API token>",
            transport=AiohttpTransport(session, headers={"X-Team": "checkout"}, timeout=10),
        ) as client:
            ...

AiohttpTransport's timeout applies to every request, in place of the session's own. Every status reaches the client as a response, even on a session created with raise_for_status=True.

An AiohttpTransport() without a session opens its own; if you create one and pass it in, close it with await transport.aclose() after the client.

Another HTTP library

A transport receives a Request (method, a Method: GET or POST, plus url, headers, params and body) and returns a Response (status, headers, body as bytes). Every status is a response: raise only when no response arrived, such as on a connection error or a timeout. The client logs the error and reports a failed request. For example, with httpx:

import httpx
from UnleashClient import Request, Response, UnleashClient

class HttpxTransport:
    def __init__(self, client: httpx.Client) -> None:
        self._client = client

    def execute(self, request: Request) -> Response:
        resp = self._client.request(
            request.method,
            request.url,
            headers=dict(request.headers),
            params=dict(request.params),
            content=request.body,
        )
        return Response(status=resp.status_code, headers=resp.headers, body=resp.content)

with httpx.Client(timeout=30) as http_client:
    client = UnleashClient(
        url="https://YOUR-API-URL",
        app_name="my-python-app",
        api_key="<API token>",
        transport=HttpxTransport(http_client),
    )

For AsyncUnleashClient, execute is async. The same with httpx.AsyncClient:

import httpx
from UnleashClient import Request, Response
from UnleashClient.clients.async_unleash_client import AsyncUnleashClient

class HttpxAsyncTransport:
    def __init__(self, client: httpx.AsyncClient) -> None:
        self._client = client

    async def execute(self, request: Request) -> Response:
        resp = await self._client.request(
            request.method,
            request.url,
            headers=dict(request.headers),
            params=dict(request.params),
            content=request.body,
        )
        return Response(status=resp.status_code, headers=resp.headers, body=resp.content)

async def main():
    async with httpx.AsyncClient(timeout=30) as http_client:
        async with AsyncUnleashClient(
            url="https://YOUR-API-URL",
            app_name="my-python-app",
            api_key="<API token>",
            transport=HttpxAsyncTransport(http_client),
        ) as client:
            ...

The response headers can be a plain dict: the client looks them up case-insensitively.

Tests without a server

A fake transport lets your tests choose the flags without a server or HTTP mocking:

import json
from UnleashClient import Method, Request, Response, UnleashClient

class FakeTransport:
    def __init__(self, *flags):
        self._features = json.dumps({"version": 1, "features": list(flags)}).encode()
        self.requests = []  # what the SDK sent

    def execute(self, request: Request) -> Response:
        self.requests.append(request)
        if request.method == Method.GET:
            return Response(status=200, headers={}, body=self._features)
        return Response(status=202, headers={}, body=b"")  # registration and metrics

def flag(name, enabled):
    return {"name": name, "enabled": enabled, "strategies": [{"name": "default", "parameters": {}}]}

def test_new_checkout_is_on(tmp_path):
    transport = FakeTransport(flag("new-checkout", True))

    with UnleashClient(
        "http://unleash.test/api", "my-app-tests",
        transport=transport, cache_directory=str(tmp_path),
    ) as client:
        assert client.is_enabled("new-checkout")

For AsyncUnleashClient, make execute async and the test a coroutine, here with pytest-asyncio:

import pytest
from UnleashClient.clients.async_unleash_client import AsyncUnleashClient


class FakeAsyncTransport(FakeTransport):
    async def execute(self, request: Request) -> Response:
        return super().execute(request)

@pytest.mark.asyncio
async def test_new_checkout_is_on(tmp_path):
    transport = FakeAsyncTransport(flag("new-checkout", True))

    async with AsyncUnleashClient(
        "http://unleash.test/api", "my-app-tests",
        transport=transport, cache_directory=str(tmp_path),
    ) as client:
        assert client.is_enabled("new-checkout")

All four examples are in examples/ and run as part of the SDK's tests.

Things to know

Don't retry in your transport. The client already retries a failed feature fetch up to 3 times, with a backoff. A requests session with a retrying adapter mounted logs a warning, because failed fetches would be retried twice over. aiohttp has no retry setting, but resends a GET once by itself when the server drops a reused connection; that's expected, and not a retry to turn off.
Don't log request headers. They include your API token.
Don't add an HTTP cache. The client already sends If-None-Match and handles 304 Not Modified. A caching session (such as CacheControl or requests-cache) turns each 304 into its cached 200, so the client reloads unchanged flags on every poll. The flags stay correct.
Create the session in the process that uses it. With a forking server, such as gunicorn with --preload or uWSGI without lazy-apps, build the session together with the client in each worker. Forked processes must not share open connections. See running in multi-process setups.
Streaming mode doesn't use the transport. With mode=StreamingOptions(), the stream connects with the SDK's own headers and a 30 second timeout. Custom transport's headers, timeout, proxy and TLS settings don't apply to it.

Running in multi-process setups

The Python SDK runs a background thread to keep feature flags in sync with the Unleash server. Some runtime environments, like WSGI servers and Celery workers, need extra setup to make sure the SDK works correctly.

WSGI

When using WSGI servers (e.g., for Flask or Django apps), be aware that:

Many WSGI setups disable threading by default. The SDK needs threads to poll for updates in the background.
Make sure to set enable-threads in your WSGI config.

When running under uWSGI with multiple processes (using the --processes option), you may need to enable the lazy-apps option. This ensures each process gets a fresh SDK instance.

See The Art of Graceful Reloading for more details.

Celery

When using the SDK in Celery tasks, make sure you initialize it inside the worker_process_init event. Otherwise, the worker may run but won’t poll for feature flag updates.

from UnleashClient import UnleashClient
from celery.signals import worker_process_init

client = UnleashClient(
    url="https://YOUR-API-URL",
    app_name="my-python-app",
    api_key="<API token>"
)

@worker_process_init.connect
def configure_workers(sender=None, conf=None, **kwargs):
    client.initialize_client()

Contributing and development

We love community input! If you’d like to report a bug, propose a feature, or improve the SDK, please read our contribution guide for how to get started.

For instructions on setting up your development environment, running tests, and publishing, see our development documentation.

License

This project is MIT licensed.

GeezerVison

Latest commit

History

Folders and files

Repository files navigation

Unleash SDK for Python

Getting Started

Install the Unleash Client in your project

Initialization

Check features

Shutdown

Usage

Context

Fallback function

Configuration options

IdentityOptions

PollingOptions

StreamingOptions

MetricsOptions

Bootstrap

Bootstrap from dict

Bootstrap from file

Bootstrap from URL

Custom strategies

Events and impression data

Impression events

Lifecycle events

Impact metrics

Counters

Gauges

Histograms

Custom cache

Custom HTTP transport

Custom requests session

Custom aiohttp session

Another HTTP library

Tests without a server

Things to know

Running in multi-process setups

WSGI

Celery

Contributing and development

License

About

Topics

Resources

Code of conduct

Contributing

Stars

Watchers

Forks

Releases

Packages

Used by