Python

1
from UnleashClient import UnleashClient
2
3
client = UnleashClient(
4
    url="https:<YOUR-API-URL>",
5
    app_name="my-python-app",
6
    custom_headers={'Authorization': '<API token>'})
7
8
client.initialize_client()

1
enabled = client.is_enabled("my_toggle")
2
print(enabled)
3
> True
4
5
variant = client.get_variant("variant_toggle")
6
print(variant)
7
> {
8
>    "name": "variant1",
9
>    "payload": {
10
>        "type": "string",
11
>        "value": "val1"
12
>        },
13
>    "enabled": True
14
> }

1
client.destroy()

1
app_context = {
2
    "userId": "test@email.com",
3
    "sessionId": "55845",
4
    "properties": {
5
        "custom-property": "some-value"
6
    }
7
}
8
9
client.is_enabled("user_id_toggle", app_context)
10
client.get_variant("variant_toggle", app_context)

1
def custom_fallback(feature_name: str, context: dict) -> bool:
2
    return True
3
4
client.is_enabled("my_toggle", fallback_function=custom_fallback)

1
from UnleashClient.cache import FileCache
2
from UnleashClient import UnleashClient
3
4
# Create and seed the cache
5
cache = FileCache("MY_CACHE")
6
cache.bootstrap_from_dict({
7
    "version": 2,
8
    "features": [
9
        {
10
            "name": "my_toggle",
11
            "enabled": True,
12
            "strategies": [{"name": "default"}],
13
        }
14
    ]
15
})
16
17
client = UnleashClient(
18
    url="https://YOUR-API-URL",
19
    app_name="my-python-app",
20
    cache=cache
21
)

1
from pathlib import Path
2
from UnleashClient.cache import FileCache
3
4
cache = FileCache("MY_CACHE")
5
cache.bootstrap_from_file(Path("/path/to/your_bootstrap.json"))
6
7
client = UnleashClient(
8
    url="https://YOUR-API-URL",
9
    app_name="my-python-app",
10
    cache=cache
11
)

1
from UnleashClient.cache import FileCache
2
3
cache = FileCache("MY_CACHE")
4
cache.bootstrap_from_url("https://your-server/bootstrap.json")
5
6
client = UnleashClient(
7
    url="https://YOUR-API-URL",
8
    app_name="my-python-app",
9
    cache=cache
10
)

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

1
## You should have a custom strategy defined in Unleash called 'EmailStrategy'
2
my_custom_strategies = {
3
    "EmailStrategy": ActiveForEmailStrategy()
4
}
5
6
client = UnleashClient(
7
8
    url="https://YOUR-API-URL",
9
    app_name="my-python-app",
10
    custom_headers={'Authorization': '<API token>'},
11
    custom_strategies = my_custom_strategies
12
)

1
from blinker import signal
2
from UnleashClient import UnleashClient
3
from UnleashClient.events import UnleashEvent
4
5
send_data = signal('send-data')
6
7
@send_data.connect
8
def receive_data(sender, **kw):
9
    print("Caught signal from %r, data %r" % (sender, kw))
10
11
def example_callback(event: UnleashEvent):
12
    send_data.send('anonymous', data=event)
13
14
client = UnleashClient(
15
    url="https://YOUR-API-URL",
16
    app_name="my-python-app",
17
    custom_headers={'Authorization': '<API token>'},
18
    event_callback=example_callback
19
)
20
client.initialize_client()
21
client.is_enabled("testFlag")

1
from blinker import signal
2
from UnleashClient import UnleashClient
3
from UnleashClient.events import UnleashEvent, UnleashEventType
4
5
send_data = signal('send-data')
6
7
@send_data.connect
8
def receive_data(sender, **kw):
9
    if kw["data"].event_type == UnleashEventType.READY:
10
        print("SDK is ready: toggles loaded from Unleash or backup")
11
    elif kw["data"].event_type == UnleashEventType.FETCHED:
12
        # Only FETCHED events have a 'features' property
13
        print("Fetched new feature flags:", kw["data"].features)
14
15
def example_callback(event: UnleashEvent):
16
    send_data.send('anonymous', data=event)
17
18
client = UnleashClient(
19
    url="https://YOUR-API-URL",
20
    app_name="my-python-app",
21
    custom_headers={'Authorization': '<API token>'},
22
    event_callback=example_callback
23
)
24
client.initialize_client()
25
client.is_enabled("testFlag")

1
from typing import Optional, Any
2
from UnleashClient.cache import BaseCache
3
from fcache.cache import FileCache as _FileCache
4
5
class CustomCache(BaseCache):
6
    # This is specific for FileCache.  Depending on the cache you're using, this may look different!
7
    def __init__(self, name: str, directory: Optional[str] = None):
8
        self._cache = _FileCache(name, app_cache_dir=directory)
9
10
    def set(self, key: str, value: Any):
11
        self._cache[key] = value
12
        self._cache.sync()
13
14
    def mset(self, data: dict):
15
        self._cache.update(data)
16
        self._cache.sync()
17
18
    def get(self, key: str, default: Optional[Any] = None):
19
        return self._cache.get(key, default)
20
21
    def exists(self, key: str):
22
        return key in self._cache
23
24
    def destroy(self):
25
        return self._cache.delete()

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

1
from UnleashClient import UnleashClient
2
from celery.signals import worker_process_init
3
4
client = UnleashClient(
5
    url="https://YOUR-API-URL",
6
    app_name="my-python-app",
7
    custom_headers={'Authorization': '<API token>'}
8
)
9
10
@worker_process_init.connect
11
def configure_workers(sender=None, conf=None, **kwargs):
12
    client.initialize_client()