traffical 0.3.1

Traffical Python SDK: deterministic experimentation and parameter resolution

Traffical Python SDK

Traffical is one control plane for experiments, feature flags, and adaptive optimization. Instead of hard-coding decisions, you expose typed parameters — numbers, strings, booleans, and JSON, not just on/off toggles — and control behavior across web, mobile, push, and backend from a single place. Parameters are resolved locally in the SDK (sub-millisecond, no network round trips at runtime), and metrics are computed warehouse-native, against data you own. Start with a feature flag, graduate it to an A/B test, and let adaptive optimization shift traffic to the winning variant — all on the same parameter, without a deploy.

This package is the official Python SDK (Python 3.10+) for Traffical, with first-class sync and asyncio clients.

Features

• Local, in-process evaluation — resolve a cached config bundle with no per-decision network call.
• Typed parameters with safe defaults — bool / string / number / JSON, each with a caller-provided fallback; every public call is fail-open (your defaults always win on errors).
• Layered experiments & targeting — Google-style layered isolation, condition/attribute segmentation, and progressive (percentage) rollouts.
• Adaptive optimization — contextual-bandit scoring evaluated client-side.
• BYO warehouse-native assignment logging — route structured assignment rows through your own pipeline (a DB, a queue, Segment, RudderStack) so assignment data never has to leave your infrastructure.
• Event tracking — exposure, decision, and custom track events, batched and flushed on a background thread (or asyncio task), with retries and an atexit safety net.
• Fork-safe by construction — works under gunicorn pre-fork, celery, and anything else that calls os.fork() after client creation.
• Pure, deterministic engine — traffical.engine is stdlib-only (no I/O, no clocks, no globals) and conformance-tested against the cross-language SDK spec.

Modes

• Bundle mode (default) — the SDK fetches a config bundle, refreshes it on a background poller, and resolves every parameter locally. No per-decision network call.
• Server mode — resolution is delegated to the Traffical edge via POST /v1/resolve. Use it when you want zero client-side evaluation logic.

Installation

pip install traffical
# or
uv add traffical

Requires Python 3.10+. The only runtime dependency is httpx.

Quick start

from traffical import TrafficalClient, TrafficalClientOptions

client = TrafficalClient(
    TrafficalClientOptions(
        api_key="sk_...",
        project_id="proj_...",
        env="production",
        org_id="org_...",
    )
)
client.wait_for_ready(timeout=5.0)

# Resolve parameters with defaults as the fallback.
params = client.get_params(
    context={"userId": "user-abc", "country": "US"},
    defaults={"ui.primaryColor": "#000000", "pricing.discount": 0},
)
color = params["ui.primaryColor"]

# decide() returns a Decision (assignments + metadata). Call track_exposure()
# when the user actually sees the treatment.
decision = client.decide({"userId": "user-abc"}, {"ui.primaryColor": "#000000"})
variant = decision.assignments["ui.primaryColor"]
client.track_exposure(decision)

# Custom analytics events, attributed to the decision. Numeric metrics go in
# the value / values options (not properties):
client.track("checkout_completed", value=49.0, decision_id=decision.decision_id)

client.close()  # flushes pending events; also usable as a context manager

TrafficalClient supports with-statement usage; close() is idempotent and flushes the event queue.

Async quick start

AsyncTrafficalClient has the same surface with awaitable get_params / decide / flush_events / close. Background work runs on the event loop; enter the context manager (or call start()) from a running loop.

import asyncio

from traffical import AsyncTrafficalClient, TrafficalClientOptions


async def main() -> None:
    async with AsyncTrafficalClient(
        TrafficalClientOptions(
            api_key="sk_...",
            project_id="proj_...",
            env="production",
            org_id="org_...",
        )
    ) as client:
        await client.wait_for_ready(timeout=5.0)
        decision = await client.decide({"userId": "user-abc"}, {"ui.primaryColor": "#000000"})
        client.track_exposure(decision)


asyncio.run(main())

Server evaluation mode

Delegate resolution to the edge worker instead of evaluating a local bundle. Each get_params() / decide() performs a POST /v1/resolve:

options = TrafficalClientOptions(
    api_key="sk_...",
    project_id="proj_...",
    env="production",
    org_id="org_...",
    evaluation_mode="server",
)

In server mode wait_for_ready() returns immediately (there is no bundle to wait for) and resolve failures fall back to your defaults.

Warehouse-native assignment logging

Pass an assignment_logger callable to route structured per-layer rows through your own pipeline. Each AssignmentLogEntry carries the stable policy_key / allocation_key used for warehouse joins, plus unit_key, layer_id, decision_id, timestamps and SDK metadata. Combine with disable_cloud_events=True to keep assignment data entirely on your own infrastructure:

from traffical import AssignmentLogEntry, TrafficalClient, TrafficalClientOptions


def log_assignment(entry: AssignmentLogEntry) -> None:
    my_warehouse.insert("assignments", entry.__dict__)  # DB, queue, CDP, ...


client = TrafficalClient(
    TrafficalClientOptions(
        api_key="sk_...",
        project_id="proj_...",
        env="production",
        org_id="org_...",
        assignment_logger=log_assignment,
        disable_cloud_events=True,  # keep assignment data on your own infra
    )
)

The logger fires on every decide() (type "decision") and track_exposure() (type "exposure"), independent of track_decisions, decision deduplication, and disable_cloud_events. Entries are deduplicated per unit:policy:allocation:type with a 1-hour TTL; set deduplicate_assignment_logger=False to receive every row.

AssignmentLogEntry.to_row() maps the entry onto a flat, warehouse-friendly snake_case row. Two field names are deliberately renamed for the warehouse convention: the propensity is emitted as propensity (not probability), and the entry id as assignment_id. Off-policy training columns (bucket, propensity, model_version, config_version) are included, and any filtered context (properties) is spread to the top level as CUPED covariates.

Options reference

TrafficalClientOptions is a frozen dataclass; construct it with keyword arguments.

Option	Default	Description
api_key, project_id, env, org_id	—	Required scoping + auth
base_url	https://sdk.traffical.io	Control-plane base URL
evaluation_mode	"bundle"	"bundle" (local) or "server" (POST /v1/resolve)
refresh_interval_seconds	60.0	Bundle refresh interval (seconds); <= 0 fetches once, no background refresh
config_timeout_seconds	10.0	HTTP timeout for config-bundle fetch (seconds)
events_timeout_seconds	10.0	HTTP timeout for event delivery (seconds)
resolve_timeout_seconds	5.0	HTTP timeout for server-mode resolve calls (seconds)
track_decisions	True	Emit decision events on decide()
decision_dedup_ttl_seconds	3600.0	Decision-event dedup window (seconds)
deduplicate_exposures	True	Session-dedup exposures per unit/policy/allocation
exposure_session_ttl_seconds	1800.0	Exposure session-dedup window (seconds; 30 min)
batch_size	10	Auto-flush threshold
flush_interval_seconds	30.0	Background flush interval (seconds)
max_event_queue_size	1000	Bounded event queue; overflow drops oldest
event_max_retries	3	Delivery retries per batch
assignment_logger	None	BYO warehouse logger, (AssignmentLogEntry) -> None
deduplicate_assignment_logger	True	Dedup logger rows per unit/policy/allocation/type
disable_cloud_events	False	Stop sending events to Traffical
local_config	None	Bootstrap/offline ConfigBundle (or raw dict)
config_source	None	Custom bundle source (FileConfigSource, InlineConfigSource, CallableConfigSource, or your own)
transport, async_transport	None	httpx transport seams (e.g. MockTransport in tests)

Fork safety (gunicorn, celery, …)

Pre-fork servers os.fork() after your app module — and your client — are created. Background threads do not survive a fork, and inherited locks can be held by threads that no longer exist. The SDK handles this automatically via os.register_at_fork:

• locks in the event queue, flusher, dedup caches, and config poller are re-armed in the child,
• dead background threads (event flusher, config poller) are dropped and restarted lazily on next use,
• each child process flushes its own events; an atexit/weakref.finalize safety net stops flushers of clients that are garbage-collected without close().

What you should know:

• Creating the client at module import (pre-fork) is fine and recommended — workers share nothing at runtime.
• Call close() (or flush_events()) in worker shutdown hooks if you need a hard delivery guarantee.
• The asyncio client is bound to the event loop it was started on; create it per process/loop, not pre-fork.

Cross-language conformance

The Python SDK shares the language-agnostic Traffical SDK spec (0.7.0) with the JS/TS and PHP SDKs: the same SHA-256 v2 (UTF-8 byte) assignment hashing, the same layered resolution engine, and the same contextual-bandit scoring. Every release is gated on the spec's deterministic conformance vectors — all resolution fixture suites (basic, conditions, conditions_omitted, contextual, contextual_boundary, contextual_gamma_zero, contextual_high_floor, edge_policies, per_layer_unit_key, empty_unit_key, numeric_unit_key, unicode, plus the entity-weight and resolve vectors) — so a given unit buckets identically on every platform.

The 0.7.0 drift-remediation vectors lock the cross-SDK behavior decisions this SDK implements: empty layer unitKey overrides skip the layer (S1), numeric unit keys stringify via ECMAScript Number::toString (S2), strict condition typing with .length paths (S3), single-event exposure with session dedup (S4), omitted relational values never match (S5), the safeGamma/effectiveFloor softmax guards (S6), and modelVersion = generatedAt ?? modelVersion (S7). Emitted exposure/decision/track payloads are also validated against events.schema.json and the events_conformance.json vectors.

Examples

Runnable scripts live in examples/: a sync quick start, an asyncio variant, and a FastAPI integration. They run offline against a local bundle file (FileConfigSource), so no API key is needed.

Development

uv sync                                  # create the venv + install dev deps
uv run pytest                            # full suite (unit + conformance)
uv run mypy src/traffical                # strict type checking
uv run ruff check                        # lint

Conformance fixtures are loaded from the sdk-spec checkout, resolved from the TRAFFICAL_SDK_SPEC_PATH environment variable and defaulting to the sibling ../sdk-spec directory:

TRAFFICAL_SDK_SPEC_PATH=/path/to/sdk-spec uv run pytest tests/conformance

See CONTRIBUTING.md for the full workflow.

License

MIT — see LICENSE.