CodeExamples

Code Examples

Complete, runnable examples for every public method on Core and Context. Each example focuses on one method or concern — combine them freely. For config field details, see Configuration; for type definitions, see Type Hints.

Sections:

Initialization

Direct config (no network)

from convert_sdk import Core, SDKConfig

core = Core(
    SDKConfig(
        data={
            "account_id": "1001",
            "project": {"id": "2002", "name": "Demo"},
            "experiences": [],
            "features": [],
            "goals": [],
        }
    )
).initialize()

assert core.is_ready
core.close()

SDK key (remote fetch)

Read the key from the environment — never hard-code credentials:

import os
from convert_sdk import Core, SDKConfig, TransportConfig

core = Core(
    SDKConfig(
        sdk_key=os.environ["CONVERT_SDK_KEY"],
        environment="production",
        transport=TransportConfig(timeout=5.0),
    )
).initialize()

assert core.is_ready
core.close()

Context-manager form

Use the context-manager form so close() is always called, even on error:

from convert_sdk import Core, SDKConfig

with Core(SDKConfig(data={"account_id": "1001", "project": {"id": "2"}, "experiences": [], "features": [], "goals": []})).initialize() as core:
    context = core.create_context("visitor-001")
    result = context.run_experience("checkout-experiment")
# core.close() called automatically on exit

Long-running service with auto-refresh

import os
from convert_sdk import Core, SDKConfig, RefreshConfig, LifecycleEvent

core = Core(
    SDKConfig(
        sdk_key=os.environ["CONVERT_SDK_KEY"],
        refresh=RefreshConfig(
            interval_seconds=300.0,     # refresh every 5 minutes
            jitter_seconds=30.0,        # avoid thundering herd across instances
            backoff_factor=2.0,         # exponential backoff on transient failures
            backoff_max_seconds=600.0,  # never back off longer than 10 minutes
        ),
    )
).initialize()

# Subscribe to config-swap events (e.g. to bust a downstream cache):
core.on(LifecycleEvent.CONFIG_UPDATED, lambda payload, error=None: print("config swapped"))

# Trigger an immediate out-of-band refresh (fire-and-forget):
core.refresh_now()

core.close()   # stops the background refresh thread

Accessing the current snapshot

core.current_config returns the current immutable ConfigSnapshot, or None if not yet initialized. It is safe to call from any thread:

snapshot = core.current_config
if snapshot is not None:
    print(snapshot.project_id, snapshot.account_id)

Creating a visitor context

core.create_context(visitor_id) returns a Context bound to the current immutable snapshot. Attributes are copied defensively:

context = core.create_context(
    "visitor-abc",
    visitor_attributes={"country": "US", "plan": "pro"},
)

# Read stored attributes back:
print(context.visitor_id)                     # "visitor-abc"
print(dict(context.visitor_attributes))       # {"country": "US", "plan": "pro"}
print(dict(context.attributes))               # same as visitor_attributes (alias)
print(dict(context.default_segments))         # {} — segments are separate

Reuse the same Context within a request. Create a new one per request (or per visitor interaction) to pick up the latest config.

Updating stored attributes

set_attributes merges new keys into the stored state and persists through the configured DataStore. Later create_context calls for the same visitor rehydrate the merged state:

context.set_attributes({"plan": "enterprise", "beta": True})
# context.visitor_attributes now includes all prior keys plus the update.

Experience evaluation

Single experience

run_experience returns ExperienceResult | None. A None result is a normal, non-exceptional outcome — audience miss, outside traffic allocation, or experience not found:

from convert_sdk import ExperienceResult

result = context.run_experience("checkout-experiment")

if result is None:
    pass   # visitor did not qualify — not an error
else:
    assert isinstance(result, ExperienceResult)
    print(result.experience_key)   # "checkout-experiment"
    print(result.experience_id)    # e.g. "100342"
    print(result.variation_key)    # e.g. "treatment" (may be None if unset in config)
    print(result.variation_id)     # e.g. "200512"
    # result.variation is a read-only mapping of the raw variation config

Per-call attribute overlay

attributes= is an ephemeral overlay for a single call. It does not mutate the context's stored visitor_attributes:

# Stored attribute: country="US"
result = context.run_experience(
    "geo-experiment",
    attributes={"country": "DE"},   # this call only; context unchanged
)
assert context.visitor_attributes["country"] == "US"

All applicable experiences

run_experiences() evaluates every active experience in the config and returns a list of results for those the visitor qualifies for. The list is empty (not None) when no experience resolves:

results = context.run_experiences()
for r in results:
    print(r.experience_key, "->", r.variation_key)

Both run_experience and run_experiences accept location_attributes= as a keyword-only argument for location-rule qualification:

result = context.run_experience(
    "landing-page-test",
    location_attributes={"url": "/pricing"},
)

Evaluate without reporting the bucketing event

Bucketing a visitor enqueues a bucketing/activation event so the visitor is counted in the experiment's exposure denominator. Pass enable_tracking=False (keyword-only, default True) to evaluate without reporting it — bucketing, sticky persistence, and audience-rule evaluation all still run, and the returned result is identical; the flag gates tracking only, never evaluation (useful for consent-denied flows or previewing a variation before applying it):

variation = context.run_experience("homepage-test", enable_tracking=False)
context.run_experiences(enable_tracking=False)

Both reporting paths are suppressed together: the outbound network enqueue and the in-process BUCKETING lifecycle event. So a subscriber registered through the event bus receives nothing for a call you explicitly asked to be silent — if you need a decision signal regardless of consent, read the returned result rather than subscribing to the event. (This is a deliberate Python/Android choice; the JavaScript, PHP, Ruby and iOS SDKs fire their bucketing lifecycle event even when the tracking switch is off, because there it sits outside the tracking path.)

Python's tracking control is per-call only — there is no global tracking switch. Leave enable_tracking at its True default for production traffic, or your experiment's exposure count goes to zero and you lose the conversion-rate denominator.

Evaluate without persisting the decision

A successful bucket is written into the visitor's sticky-bucketing map — in memory, and through the configured DataStore when one is set — so a later call for the same experience returns the stored decision instead of re-hashing. Pass enable_storage=False (keyword-only, default True, JS parity with BucketingAttributes.enableStorage) to skip that write for one call:

# Evaluate, report the exposure, but leave no stored decision behind:
context.run_experience("homepage-test", enable_storage=False)

# The two flags are independent — combine them for a fully silent evaluation:
context.run_experiences(enable_tracking=False, enable_storage=False)

enable_storage gates only the bucketing-decision write. Unrelated persistence — set_attributes, set_segments, run_custom_segments — is untouched, and the returned result is identical either way. run_feature / run_features have no such flag: they read the sticky map so a feature stays consistent with an already-served decision, but never write to it.

A sticky read-back is also what makes mutually exclusive experiments work across requests — see Mutually Exclusive Experiments.

Preview links

A preview link renders one specific variation of one experiment for whoever opens it, bypassing audiences, status, traffic, and bucketing — and recording nothing. Your application reads the link parameter and hands the pair to the context before the experience renders.

parse_preview_param is a pure top-level helper: it parses "{experience_id}.{variation_id}" and returns a (str, str) tuple, or None for anything malformed (missing dot, more than one dot, an empty or non-numeric segment). It never raises, so the guard falls straight through to normal behavior:

from convert_sdk import parse_preview_param

parse_preview_param("100342.200512")   # ("100342", "200512")
parse_preview_param("100342")          # None — no dot
parse_preview_param("1.2.3")           # None — more than one dot
parse_preview_param("abc.200512")      # None — non-numeric segment

Context.set_preview(experience_id, variation_id) applies the forced decision. Note it takes ids, not keys — the same numeric ids the link carries:

# Flask-style: the query parameter, backed by a short-lived cookie so the
# preview survives navigation.
raw = (
    request.args.get("convert_preview")
    or request.cookies.get("convert_preview")
    or ""
)
pair = parse_preview_param(raw)
if pair:
    context.set_preview(pair[0], pair[1])

variation = context.run_experience("checkout-experiment")   # forced variation

Resolution is eager: set_preview resolves the experience and validates the variation id at call time, not at the next run_experience. If the id is not in the loaded snapshot, the SDK fetches just that one experience on demand — so a preview link works on a normal production build for a draft or paused experiment, with no debug token. That on-demand fetch needs a remote-config (sdk_key) Core whose transport supports it, which the bundled transport does; see Extending § supporting preview fetch.

Two behaviors worth designing around:

Inert on bad input. An empty id, an experience that resolves neither locally nor remotely, a failed fetch, or an unknown variation id all degrade to a logged warning with no forcing stored. The call never raises and the visitor sees the normal experience.
Zero trace, whole context. While a preview is active, no bucketing events, no conversions, and no DataStore writes happen — not just for the forced experiment but for every other experience evaluated on that same context. track_conversion returns an untracked ConversionResult (tracked is False). The suppression is per-context: a concurrent non-preview Context sharing the same Core is unaffected.

For the debug-token path — loading every draft and paused experience at once for in-code QA — see Configuration § debug_token and the QA & Preview guide.

Feature evaluation

run_feature resolves a feature flag and its typed variables from the visitor's selected variation. It returns FeatureResult | None — None when the feature is not declared, the visitor is not bucketed into any variation carrying the feature, or the feature is otherwise unavailable:

from convert_sdk import FeatureStatus

feature = context.run_feature("checkout-banner")

if feature is None:
    pass   # feature unavailable or visitor not bucketed into it
else:
    assert feature.status is FeatureStatus.ENABLED
    print(feature.feature_key)     # "checkout-banner"
    print(feature.feature_id)      # the feature's config id
    print(feature.experience_key)  # which experience supplied the change
    print(feature.variation_key)   # the selected variation's key

    # Variables are type-cast from the feature's declared variable types:
    headline = feature.variables.get("headline")   # str
    enabled  = feature.variables.get("enabled")    # bool
    max_items = feature.variables.get("max_items") # int

All applicable features

run_features() resolves every declared feature the visitor buckets into, returning a list (empty if none):

for f in context.run_features():
    print(f.feature_key, f.status.value, dict(f.variables))

Both run_feature and run_features accept attributes= and location_attributes= overlays, identical to the experience methods.

Segments

Default segments

Default segments feed reporting and conversion attribution. They are kept strictly separate from visitor_attributes — do not mix them. Use set_segments to associate default segments with the visitor:

context.set_segments({"customerType": "vip", "loyalty_tier": "gold"})
print(dict(context.default_segments))
# {"customerType": "vip", "loyalty_tier": "gold"}

Like set_attributes, set_segments merges into the stored state and persists through the DataStore. The next create_context for the same visitor rehydrates both attributes and segments.

Custom segment evaluation

run_custom_segments evaluates named segment rules against the visitor and records newly matched segment IDs in the visitor's default-segment state. Returns a typed CustomSegmentsResult — never None, never raises on a miss:

from convert_sdk import CustomSegmentsResult

# rule_data is an ephemeral per-call overlay — never written back to
# visitor_attributes:
result = context.run_custom_segments(
    ["us-visitors", "premium-plan"],
    rule_data={"country": "US", "plan": "pro"},
)

assert isinstance(result, CustomSegmentsResult)
# result.matched_segment_ids: tuple[str, ...] — IDs newly matched THIS call
# result.matched: bool convenience property
print(result.matched_segment_ids)   # e.g. ("seg_us_abc",)
print(result.matched)               # True

# A no-match is a typed empty result, not an exception:
miss = context.run_custom_segments(["us-visitors"], rule_data={"country": "DE"})
assert miss.matched is False
assert miss.matched_segment_ids == ()

Already-matched segment IDs are not re-added on subsequent calls.

Conversion tracking

track_conversion is synchronous and lightweight — it resolves the goal, deduplicates, and appends to the in-process queue. No network call happens. It always returns a typed ConversionResult:

from convert_sdk import ConversionStatus

result = context.track_conversion("purchase_completed")

# Inspect the outcome without exceptions:
if result.status is ConversionStatus.QUEUED:
    print("queued:", result.tracked)    # True
elif result.status is ConversionStatus.DEDUPLICATED:
    print("duplicate:", result.reason)  # "deduplicated"
elif result.status is ConversionStatus.GOAL_NOT_FOUND:
    print("missing:", result.reason)    # "goal_not_found"

`ConversionStatus`	`tracked`	`reason`	Meaning
`QUEUED`	`True`	`None`	Goal resolved; event enqueued.
`DEDUPLICATED`	`False`	`"deduplicated"`	Same `(visitor, goal)` already tracked; no second event.
`GOAL_NOT_FOUND`	`False`	`"goal_not_found"`	Goal key absent from loaded config; typed non-exception outcome.

Conversion with revenue

result = context.track_conversion(
    "purchase_completed",
    revenue=49.99,
)
assert result.tracked is True

Revenue with custom conversion data

conversion_data values must be JSON primitives (int, float, str). Objects, lists, and booleans raise ConversionDataError (programmer misuse):

result = context.track_conversion(
    "purchase_completed",
    revenue=149.00,
    conversion_data={
        "transaction_id": "txn-abc-123",
        "items_count": 3,
    },
)

Allowing repeated transactions (`force_multiple`)

By default, deduplication is keyed by (visitor_id, goal_id) — a differing revenue or conversion_data does not defeat it. Use force_multiple=True to re-track an already-tracked goal (e.g. recording multiple revenue transactions in one session):

# First purchase:
context.track_conversion("purchase_completed", revenue=49.99)

# Second purchase in the same session:
again = context.track_conversion(
    "purchase_completed",
    revenue=29.99,
    conversion_data={"transaction_id": "txn-2"},
    force_multiple=True,    # overrides default-mode dedup
)
assert again.tracked is True

Attribution context

A tracked conversion carries the visitor's attribution context at conversion time: active default segments (set_segments) and active variation/bucketing assignments. Call set_segments and run_experience/run_feature before track_conversion so the conversion is attributed correctly.

Queue control and flush

track_conversion never sends to the network — it appends to an in-process batch queue. Use core.flush() to deliver queued events explicitly:

context.track_conversion("purchase_completed", revenue=49.99)
context.track_conversion("add_to_cart")

core.flush()   # delivers all queued events and clears the queue
core.flush()   # safe no-op on an empty queue — no transport call, no error

Batch-size automatic release

Set batch_size so the queue self-releases when it fills. This bounds in-process memory and latency without per-call flush() calls:

from convert_sdk import Core, SDKConfig

core = Core(SDKConfig(data=project_config, batch_size=25)).initialize()
# The queue releases automatically once 25 events accumulate.

Periodic auto-flush (opt-in)

auto_flush_interval_ms starts a daemonic timer for long-lived server processes. Leave as None for Lambda, cron jobs, and other short-lived runtimes:

from convert_sdk import Core, SDKConfig

# Flush every 5 seconds in a long-lived background worker:
core = Core(SDKConfig(data=project_config, auto_flush_interval_ms=5000)).initialize()

Recommended flush points by runtime

Runtime	Recommended flush point
Django / Flask (WSGI)	Response middleware `process_response()` hook
FastAPI / Starlette (ASGI)	Background task or `finally` in the route handler
AWS Lambda	End of handler, before `return`
CLI / script	`finally` block after main logic
Long-running server	Context manager (`with Core(...).initialize() as core:`)

Lifecycle event subscription

Register handlers with core.on(event, handler). Handlers receive (payload, error=None) — error is non-None only when the release failed. Handlers that raise are isolated and logged; they never affect evaluation or delivery. Safe to call before initialize():

from convert_sdk import LifecycleEvent
from convert_sdk.events import QueueReleasedPayload, ConversionEventPayload

def on_queue_released(payload: QueueReleasedPayload, error=None) -> None:
    if error is not None:
        print(f"delivery failed: status={payload.status_code} retries={payload.retry_attempts}")
    else:
        print(
            f"delivered batch: reason={payload.reason} "
            f"events={payload.event_count} visitors={payload.visitor_count}"
        )

def on_conversion(payload: ConversionEventPayload, error=None) -> None:
    print(f"conversion queued: visitor={payload.visitor_id} goal={payload.goal_key}")

def on_config_updated(payload, error=None) -> None:
    print(f"config swapped: project={payload['project_id']} entities={payload['entity_counts']}")

core.on(LifecycleEvent.API_QUEUE_RELEASED, on_queue_released)
core.on(LifecycleEvent.CONVERSION, on_conversion)
core.on(LifecycleEvent.CONFIG_UPDATED, on_config_updated)

Available lifecycle events

Event	When fired	Payload type
`LifecycleEvent.READY`	`initialize()` succeeds	internal
`LifecycleEvent.CONFIG_UPDATED`	Background refresh swaps in a new snapshot, or on first successful init	`dict` with `account_id`, `project_id`, `entity_counts`
`LifecycleEvent.CONVERSION`	A conversion event is enqueued (not on dedup or miss)	`ConversionEventPayload`
`LifecycleEvent.API_QUEUE_RELEASED`	The tracking queue is released (success or delivery failure)	`QueueReleasedPayload`
`LifecycleEvent.DATA_STORE_QUEUE_RELEASED`	(JS parity; out of MVP scope)	—

QueueReleasedPayload fields:

Field	Type	Meaning
`reason`	`ReleaseReason`	Why the release fired: `EXPLICIT`, `BATCH_SIZE`, `TIMER`, or `ATEXIT`
`batch_size`	`int`	Number of events delivered in this batch
`visitor_count`	`int`	Distinct visitor IDs in this batch
`event_count`	`int`	Total event records delivered
`status_code`	`int \| None`	HTTP status on failure; `None` on success
`retry_attempts`	`int \| None`	Exhausted retry count; `None` or `0` if the adapter does not retry

Import QueueReleasedPayload and ConversionEventPayload from convert_sdk.events.

Diagnostics

When you need to know why an experience or feature did not resolve — without raising — use the diagnose_* methods. They return typed diagnostic objects and never raise on a normal miss:

from convert_sdk import DiagnosticReason

# Experience diagnostic:
diag = context.diagnose_experience("checkout-experiment")
print(diag.resolved)           # bool — True only when RESOLVED
print(diag.reason)             # DiagnosticReason enum, e.g. AUDIENCE_MISMATCH
print(diag.reason.value)       # str wire value, e.g. "audience_mismatch"
print(diag.message)            # human-readable description
print(dict(diag.details))      # allowlist-safe detail map

All diagnose methods

# Feature — RESOLVED, FEATURE_NOT_FOUND, or FEATURE_NOT_IN_SELECTED_VARIATIONS:
feat_diag = context.diagnose_feature("checkout-banner")

# Goal — RESOLVED or GOAL_NOT_FOUND:
goal_diag = context.diagnose_goal("purchase_completed")

# Config-entity lookup — RESOLVED, ENTITY_NOT_FOUND, or PROJECT_MAPPING_REQUIRED:
entity_diag = context.diagnose_entity("experience", "checkout-experiment")

DiagnosticReason values

Value	String	Produced by
`RESOLVED`	`"resolved"`	All `diagnose_*` — visitor qualified and the entity resolved
`AUDIENCE_MISMATCH`	`"audience_mismatch"`	`diagnose_experience` — experience exists but visitor did not qualify
`EXPERIENCE_NOT_FOUND`	`"experience_not_found"`	`diagnose_experience` — key not in config
`FEATURE_NOT_FOUND`	`"feature_not_found"`	`diagnose_feature` — key not in config
`FEATURE_NOT_IN_SELECTED_VARIATIONS`	`"feature_not_in_selected_variations"`	`diagnose_feature` — feature declared but visitor's variation does not carry it
`GOAL_NOT_FOUND`	`"goal_not_found"`	`diagnose_goal` — key not in config
`ENTITY_NOT_FOUND`	`"entity_not_found"`	`diagnose_entity` — key/id not found or wrong `entity_type`
`PROJECT_MAPPING_REQUIRED`	`"project_mapping_required"`	`diagnose_entity` — loaded config has no project mapping

Miss-path reasons are also emitted as structured log records through the diagnostic logger ("convert_sdk" namespace). See Diagnostics for the full log-record field reference.

Graceful shutdown

core.close() stops the background refresh thread (if running), cancels the periodic-flush timer (if set), and closes the httpx transport if Core created it. It does not perform a final flush:

# Explicit call:
core.flush()   # deliver any remaining queued events first
core.close()   # then release resources

# Or use the context-manager form (close is called automatically on exit):
with Core(SDKConfig(data=project_config)).initialize() as core:
    ...

close() is idempotent. Calling it multiple times is safe.

What to read next

Configuration — every option of every config dataclass
Type Hints — dataclasses, Protocols, and enums for every result type
Diagnostics — structured log fields, error codes, support workflow
Extending — custom Transport, DataStore, event-bus observers
Running Experiences — detailed experience-evaluation guide
Running Features — feature-flag resolution and typed variables
Tracking Conversions — deduplication, attribution, payload reference

Getting Started

Python SDK

Migration

Core Concepts

How-To Guides

Edge & Integrations

LLM / AI Testing

Maintainers

Release Process

Uh oh!

CodeExamples

Code Examples

Initialization

Direct config (no network)

SDK key (remote fetch)

Context-manager form

Long-running service with auto-refresh

Accessing the current snapshot

Creating a visitor context

Updating stored attributes

Experience evaluation

Single experience

Per-call attribute overlay

All applicable experiences

Evaluate without reporting the bucketing event

Evaluate without persisting the decision

Preview links

Feature evaluation

All applicable features

Segments

Default segments

Custom segment evaluation

Conversion tracking

Conversion with revenue

Revenue with custom conversion data

Allowing repeated transactions (force_multiple)

Attribution context

Queue control and flush

Batch-size automatic release

Periodic auto-flush (opt-in)

Recommended flush points by runtime

Lifecycle event subscription

Available lifecycle events

Diagnostics

All diagnose methods

DiagnosticReason values

Graceful shutdown

What to read next

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally

Allowing repeated transactions (`force_multiple`)