Metadata-Version: 2.4
Name: cloudflare-flagship
Version: 0.4.0
Summary: OpenFeature provider for Cloudflare Flagship feature flags.
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: cachetools>=5.0.0
Requires-Dist: httpx>=0.28.1
Requires-Dist: openfeature-sdk>=0.9.0
Requires-Python: >=3.10
Project-URL: Homepage, https://github.com/cloudflare/flagship
Project-URL: Repository, https://github.com/cloudflare/flagship
Project-URL: Changelog, https://github.com/cloudflare/flagship/blob/main/sdks/typescript/CHANGELOG.md
Description-Content-Type: text/markdown

# cloudflare-flagship

[![PyPI version](https://img.shields.io/pypi/v/cloudflare-flagship.svg)](https://pypi.org/project/cloudflare-flagship/)
[![Python versions](https://img.shields.io/pypi/pyversions/cloudflare-flagship.svg)](https://pypi.org/project/cloudflare-flagship/)
[![license](https://img.shields.io/pypi/l/cloudflare-flagship.svg)](LICENSE)

Flagship is a globally distributed, low-latency feature flag platform built entirely on Cloudflare. This package is the Python SDK — an [OpenFeature](https://openfeature.dev)-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

```sh
# uv
uv add cloudflare-flagship

# pip
pip install cloudflare-flagship
```

## Quick start

```python
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`](examples/server.py) for a full synchronous example and [`examples/async_server.py`](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`.

```python
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:

```python
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.

```python
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):

```python
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:

```python
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:

```python
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

```python
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

```python
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

```sh
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](LICENSE)
