Python SDK

Install#

$ pip install darkship
$ poetry add darkship
$ uv add darkship
$ pip install "darkship[openfeature]"   # adds the OpenFeature provider

Initialise#

import os
import darkship

ds = darkship.init(
    sdk_key=os.environ["DARKSHIP_SDK_KEY"],   # ds_srv_prod_...
)                                               # host: DARKSHIP_BASE_URL or base_url=...
ds.on("error", lambda err: log.warning("darkship: %s", err))
ready = ds.wait_for_ready(timeout=3.0)          # True or False, never raises

Create one client per process and share it. It owns a polling thread and an in-memory ruleset; creating one per request would multiply both.

Init options#

Frameworks and process models#

from contextlib import asynccontextmanager
from fastapi import FastAPI
import darkship, os

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.ds = darkship.init(sdk_key=os.environ["DARKSHIP_SDK_KEY"])
    app.state.ds.wait_for_ready(timeout=3.0)
    yield
    app.state.ds.close(timeout=2.0)       # flushes counters

app = FastAPI(lifespan=lifespan)

from django.apps import AppConfig
import darkship, os

class CoreConfig(AppConfig):
    name = "core"
    ds = None

    def ready(self):                      # runs once per worker process
        CoreConfig.ds = darkship.init(sdk_key=os.environ["DARKSHIP_SDK_KEY"])

import darkship

with darkship.init(sdk_key=KEY) as ds:      # close() on exit
    ds.wait_for_ready(timeout=3.0)
    if ds.bool_variation("nightly-reindex", {"key": "job"}, False):
        reindex()

Evaluate flags#

ctx = {"key": "usr_marta", "plan": "pro", "country": "BR", "orgId": "org_acme"}

on      = ds.bool_variation("new-checkout-flow", ctx, False)
banner  = ds.string_variation("promo-banner-text", ctx, "Welcome back")
retries = ds.int_variation("max-retries", ctx, 3)
ratio   = ds.float_variation("sample-ratio", ctx, 0.1)
limits  = ds.json_variation("rate-limits", ctx, {"rps": 10, "burst": 20})
cutover = ds.date_variation("cutover-at", ctx, None)    # aware datetime in UTC

d = ds.get_boolean_details("new-checkout-flow", ctx, False)
# EvaluationDetail(value=True, variation='on', reason='FALLTHROUGH', error_code=None, bucket=4127)

Also: all_flags(ctx) returns {flag_key: EvaluationDetail}.

Reasons are OFF, TARGET_MATCH, RULE_MATCH, FALLTHROUGH, NOT_READY and ERROR; error codes are FLAG_NOT_FOUND, TYPE_MISMATCH, PROVIDER_NOT_READY, MALFORMED_FLAG, INVALID_CONTEXT and GENERAL. A percentage rollout adds bucket (0 to 9999) to the detail; there is no rollout reason. See the reason table. The evaluation is the same algorithm as in JavaScript and on the evaluation service, verified against shared conformance fixtures. A user_id bucketed in Python lands in the same bucket as in Node.

Context#

A dict with a required "key" (the user id) and optional "email", "name" and "groups" (a list of strings). Target lists match by key, email (normalised, case-insensitive) or groups (any element). Values may be str, int, float, bool or lists of strings. Missing attributes never match positive operators and nothing raises. Never put secrets in a context; in server mode it stays in your process.

ctx = {
    "key": "usr_marta",
    "email": "Marta@Acme.com",          # matches ana@acme.com style lists, any casing
    "groups": ["beta-testers", "acme-staff"],
    "plan": "pro",
}
on = ds.bool_variation("ai-search-beta", ctx, False)

Events#

ds.on("ready",  lambda e: log.info("ruleset %s", e["version"]))
ds.on("update", lambda e: log.info("changed: %s", e["changed"]))
ds.on("error",  lambda e: log.warning("%s: %s", e.code, e))

Handlers run on the SDK's thread, so keep them short and thread-safe.

Telemetry and privacy#

Identical to the other SDKs: per flag × variation × one-minute counters, flushed on an interval and at close(); no context and no user-level data. The field list is in the JavaScript SDK page. The sdk field reads python/<version>.

Errors and defaults#

Evaluation methods do not raise. A missing flag, a type mismatch or a not-ready client all return the default. Programmer errors at init (ValueError for a missing or non-server key) are the only exceptions. wait_for_ready returns False on timeout. If the network drops after a ruleset was loaded, evaluation continues from it and an error event is emitted on each failed refresh, with exponential backoff and jitter.

OpenFeature#

Install the extra and register the provider. It wraps the native client (pass your own with DarkshipProvider(ds)), so the two paths cannot diverge.

from openfeature import api
from openfeature.evaluation_context import EvaluationContext
from darkship.openfeature import DarkshipProvider

api.set_provider_and_wait(DarkshipProvider(sdk_key=os.environ["DARKSHIP_SDK_KEY"]))
client = api.get_client()
client.get_boolean_value("new-checkout-flow", False, EvaluationContext("usr_marta", {"plan": "pro"}))

targeting_key is the darkship key; other attributes are targeting attributes. Reasons map to STATIC (off), TARGETING_MATCH, SPLIT (a percentage rollout decided), DEFAULT (fallthrough) and ERROR. Date flags resolve as an ISO-8601 string through resolve_object_details. Provider events: READY, CONFIGURATION_CHANGED, ERROR.

Shutdown#

import atexit
atexit.register(lambda: ds.close(timeout=2.0))   # stops the thread, flushes counters (best effort)

Testing your code#

Unit tests shouldn't hit the network. Pass a bootstrap ruleset and disable the poller and telemetry:

import pytest, darkship

@pytest.fixture
def ds():
    client = darkship.init(
        sdk_key="ds_srv_test_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
        bootstrap=FIXTURE_RULESET, offline=True, telemetry=False,
    )
    yield client
    client.close()

Versions#

The package is at 0.1.0. Until 1.0, minor versions may include breaking changes, listed in the changelog. Python 3.10 or newer; the one runtime dependency is httpx. The evaluator is verified against the shared language-neutral conformance vectors.