CodeExamples

Code Examples

Complete, copy-paste examples for every public method on Core and Context. For the underlying types, see Type Hints. For init options, see Configuration.

Sections on this page:

1. Initialization
2. Visitor Context
3. Evaluation — run_experience, run_experiences, run_feature, run_features
4. Segments
5. Tracking — track_conversion, deduplication, revenue
6. Queue Control — core.flush, lifecycle events, batching
7. Diagnostics — diagnose_* for non-raising lookups

Initialization

SDK key mode (network fetch)

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

Direct config mode (no network)

from convert_sdk import Core, SDKConfig

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

Long-running service with auto-refresh

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

core = Core(
    SDKConfig(
        sdk_key=os.environ["CONVERT_SDK_KEY"],
        refresh=RefreshConfig(
            interval_seconds=300.0,
            jitter_seconds=30.0,
            backoff_factor=2.0,
            backoff_max_seconds=600.0,
        ),
    )
).initialize()

core.on(LifecycleEvent.CONFIG_UPDATED, lambda payload, error=None: my_cache.invalidate())

# Force a refresh on demand:
core.refresh_now()   # fire-and-forget

# Graceful shutdown:
core.close()

Context manager form

from convert_sdk import Core, SDKConfig

with Core(SDKConfig(data=project_config)).initialize() as core:
    context = core.create_context("visitor-1")
    ...
# core.close() called automatically on exit

Visitor Context

context = core.create_context(
    "visitor-abc123",
    visitor_attributes={"tier": "premium", "country": "US"},
)

The second argument is visitor_attributes (keyword or positional). Attributes are copied defensively — later mutations to the dict you pass never affect the context.

Update stored attributes or segments after creation:

context.set_attributes({"new_attr": "value"})   # merges into stored attributes
context.set_segments({"loyalty_tier": "gold"})   # merges into default segments (kept separate)

Context is reusable — call multiple evaluation methods on the same instance within a request lifecycle.

Evaluation

Single experience

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

if result is None:
    # visitor did not qualify (audience miss, or outside traffic)
    pass
else:
    print(result.experience_key, result.variation_key, result.variation_id)

run_experience() returns ExperienceResult | None. A None result is a normal non-exceptional outcome — see Type Hints — ExperienceResult.

Per-evaluation attribute overlay

# attributes= is an ephemeral per-call overlay; context.visitor_attributes is unchanged
result = context.run_experience(
    "beta-program",
    attributes={"beta_opt_in": True},
)

All applicable experiences

results = context.run_experiences()
# results: list[ExperienceResult], empty list when no match
for r in results:
    print(r.experience_key, r.variation_key)

Single feature

from convert_sdk import FeatureStatus

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

if feature is None:
    pass  # feature disabled or visitor not in any backing experience
elif feature.status == FeatureStatus.ENABLED:
    headline = feature.variables.get("headline")
    print(headline)

All applicable features

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

Segments

Default segments

Default segments feed reporting and conversion attribution. They are kept strictly separate from visitor_attributes.

context.set_segments({"customerType": "vip"})

Custom segment evaluation

from convert_sdk import CustomSegmentsResult

result = context.run_custom_segments(
    ["segment-premium-eu", "segment-mobile"],
    rule_data={"device": "mobile"},
)
# result: CustomSegmentsResult
# result.matched_segment_ids: tuple[str, ...] — only the IDs matched this call
# result.matched: bool

Tracking

Basic conversion

from convert_sdk import ConversionStatus

result = context.track_conversion("purchase_completed")

if result.status is ConversionStatus.QUEUED:
    print("tracked:", result.tracked)        # True
elif result.status is ConversionStatus.GOAL_NOT_FOUND:
    print("goal missing:", result.reason)    # "goal_not_found"
elif result.status is ConversionStatus.DEDUPLICATED:
    print("duplicate:", result.reason)       # "deduplicated"

Conversion with revenue data

result = context.track_conversion(
    "purchase_completed",
    revenue=49.99,
    conversion_data={
        "transaction_id": "txn-abc-123",
        "products_count": 2,
    },
)

conversion_data accepts int, float, and str values. Objects, lists, and booleans raise ConversionDataError (programmer misuse, not a normal outcome).

Allowing repeat transactions

result = context.track_conversion(
    "purchase_completed",
    revenue=29.99,
    force_multiple=True,  # overrides default-mode dedup
)

Handling a missing goal via ConversionResult

result = context.track_conversion("possibly-missing-goal")
if not result.tracked:
    print(result.reason)   # "goal_not_found" or "deduplicated"

Queue Control

The SDK queues conversion events in-process and delivers them in HTTP POST batches only when you explicitly call core.flush().

Explicit flush

core.flush()   # delivers queued events and clears the queue; no-op if empty

When to flush

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

# Django middleware example
class ConvertFlushMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self.get_response(request)
        if hasattr(request, "convert_core"):
            request.convert_core.flush()
        return response

Batch size

Configure batch_size to trigger an automatic release when the queue fills:

from convert_sdk import Core, SDKConfig

core = Core(
    SDKConfig(
        data=project_config,
        batch_size=25,      # auto-release when 25 events accumulate
    )
).initialize()

Periodic auto-flush (opt-in)

from convert_sdk import Core, SDKConfig

# Flush every 2 seconds in a long-lived server process:
core = Core(
    SDKConfig(
        data=project_config,
        auto_flush_interval_ms=2000,
    )
).initialize()

Lifecycle events for queue observability

Subscribe to lifecycle events on the Core instance to observe queue activity:

from convert_sdk import LifecycleEvent
from convert_sdk.events import QueueReleasedPayload

def on_released(payload: QueueReleasedPayload, error=None) -> None:
    if error is not None:
        print("DELIVERY FAILURE:", error)
    else:
        print(f"delivered {payload.batch_size} events, reason={payload.reason}")

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

core.on(LifecycleEvent.API_QUEUE_RELEASED, on_released)
core.on(LifecycleEvent.CONVERSION, on_conversion)

Event	Fired when
CONVERSION	A conversion event is created and enqueued
API_QUEUE_RELEASED	The tracking queue is released (success or failure)
CONFIG_UPDATED	Background refresh swapped in a new snapshot

Diagnostics

When you need to know why a visitor wasn't bucketed (or a goal wasn't found), use the diagnose_* variants. They return the full decision record without raising.

from convert_sdk import DiagnosticReason

diag = context.diagnose_experience("checkout-experiment")
print(diag.resolved)    # bool
print(diag.reason)      # DiagnosticReason enum value
print(diag.message)     # human-readable description
print(dict(diag.details))

feat_diag = context.diagnose_feature("checkout-banner")
goal_diag = context.diagnose_goal("purchase_completed")
entity_diag = context.diagnose_entity("experience", "checkout-experiment")

The full reason-code reference and field-by-field breakdown live in Diagnostics and Type Hints.

Graceful shutdown

# Application shutdown hook
core.close()  # stops the refresh thread; idempotent

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

What to read next

• Configuration — every option of every config dataclass
• Type Hints — dataclasses, Protocols, and enums
• Diagnostics — log structure, error codes, support workflow
• Extending — custom transport, storage, and event-bus