Metadata-Version: 2.4
Name: traffical
Version: 0.3.0
Summary: Traffical Python SDK: deterministic experimentation and parameter resolution
Project-URL: Homepage, https://traffical.io
Project-URL: Repository, https://github.com/traffical/python-sdk
License-Expression: MIT
License-File: LICENSE
Keywords: ab-testing,experimentation,feature-flags,traffical
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
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-Python: >=3.10
Requires-Dist: httpx<1.0,>=0.27
Description-Content-Type: text/markdown

# Traffical Python SDK

[Traffical](https://traffical.io) 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

```bash
pip install traffical
# or
uv add traffical
```

Requires Python 3.10+. The only runtime dependency is [httpx](https://www.python-httpx.org/).

## Quick start

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

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

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

```python
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](https://github.com/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/](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

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

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

See [CONTRIBUTING.md](CONTRIBUTING.md) for the full workflow.

## License

MIT — see [LICENSE](LICENSE).
