Confidence OpenFeature Provider for Python

A high-performance OpenFeature provider for Confidence feature flags that evaluates flags locally for minimal latency.

Features

• Local Resolution: Evaluates feature flags locally using WebAssembly (WASM)
• Low Latency: No network calls during flag evaluation
• Automatic Sync: Periodically syncs flag configurations from Confidence
• Exposure Logging: Fully supported exposure logging (and other resolve analytics)
• Event Tracking: Send custom events via the OpenFeature track() API
• OpenFeature Compatible: Works with the standard OpenFeature SDK

Requirements

• Python 3.10+
• OpenFeature SDK 0.10.0+

Installation

pip install confidence-openfeature-provider

Getting Your Credentials

You'll need a client secret from Confidence to use this provider.

📖 See the Integration Guide: Getting Your Credentials for step-by-step instructions on:

• How to navigate the Confidence dashboard
• Creating a Backend integration
• Creating a test flag for verification
• Best practices for credential storage

Encryption

The provider supports encrypting the flag state to protect your flag rules and targeting segments at rest and in transit. The state is decrypted only when loaded into the resolver.

📖 See the Integration Guide: Encryption for background and migration details.

Pass the encryption key when creating the provider:

provider = ConfidenceProvider(
    client_secret="your-client-secret",
    encryption_key="your-encryption-key",  # Get from Confidence Admin view
)

The encryption key is available in the Confidence Admin view, next to your client credentials.

⚠️ Upcoming change: Encryption will be made mandatory in a future SDK release. We will communicate a timeline and migration path before legacy provider versions are affected. We strongly recommend enabling it now.

Quick Start

from openfeature import api
from openfeature.evaluation_context import EvaluationContext
from confidence import ConfidenceProvider

# Create and register the provider
provider = ConfidenceProvider(client_secret="your-client-secret")
api.set_provider_and_wait(provider)

# Get a client
client = api.get_client()

# Create evaluation context with user attributes for targeting
context = EvaluationContext(
    targeting_key="user-123",
    attributes={
        "country": "US",
        "plan": "premium",
    }
)

# Evaluate a flag
enabled = client.get_boolean_value("test-flag.enabled", default_value=False, evaluation_context=context)
print(f"Flag value: {enabled}")

# Don't forget to shutdown when your application exits (see Shutdown section)

Evaluation Context

The evaluation context contains information about the user/session being evaluated for targeting and A/B testing.

Python Examples

from openfeature.evaluation_context import EvaluationContext

# Simple attributes
context = EvaluationContext(
    targeting_key="user-123",
    attributes={
        "country": "US",
        "plan": "premium",
        "age": 25,
    }
)

Error Handling

The provider uses a default value fallback pattern - when evaluation fails, it returns your specified default value instead of throwing an error.

📖 See the Integration Guide: Error Handling for:

• Common failure scenarios
• Error codes and meanings
• Production best practices
• Monitoring recommendations

Python Examples

# The provider returns the default value on errors
enabled = client.get_boolean_value("my-flag.enabled", default_value=False, evaluation_context=context)
# enabled will be False if evaluation failed

# For detailed error information, use get_boolean_details()
details = client.get_boolean_details("my-flag.enabled", default_value=False, evaluation_context=context)
if details.error_code:
    print(f"Flag evaluation error: {details.error_message}")
    print(f"Reason: {details.reason}")

Shutdown

Important: To ensure proper cleanup and flushing of exposure logs, you should call shutdown() on the provider when your application exits.

from openfeature import api
# Shutdown the provider to flush logs and clean up resources
api.shutdown()

Configuration

provider = ConfidenceProvider(
    client_secret="your-client-secret",
    state_poll_interval=30.0,  # How often to poll for state updates (seconds)
    log_poll_interval=10.0,    # How often to flush logs (seconds)
)

Configuration Options

• client_secret (str, required): The Confidence client secret for authentication.
• encryption_key (str, optional): Encryption key for decrypting the flag state. Found in the Confidence Admin view. Will be required in a future release.
• state_poll_interval (float, optional): Interval in seconds between state polling updates. Defaults to 30.0.
• log_poll_interval (float, optional): Interval in seconds for sending evaluation logs. Defaults to 10.0.
• use_remote_materialization_store (bool, optional): Enable remote materialization storage. Defaults to False.
• grpc_channel (grpc.Channel, optional): Custom gRPC channel for flag log shipping. When not provided, the default channel retries flag log writes on transient failures (3 attempts with exponential backoff on UNAVAILABLE). If you provide your own channel, configure retry via gRPC service config to get the same behavior.

Materializations

The provider supports materializations for two key use cases:

1. Sticky Assignments: Maintain consistent variant assignments across evaluations even when targeting attributes change.
2. Custom Targeting via Materialized Segments: Efficiently target precomputed sets of identifiers from datasets.

Default Behavior

By default, materializations are not supported. If a flag requires materialization data, the evaluation will return the default value.

Remote Materialization Store

Enable remote materialization storage to have Confidence manage materialization data server-side:

provider = ConfidenceProvider(
    client_secret="your-client-secret",
    use_remote_materialization_store=True,
)

⚠️ Important Performance Impact: This option adds network calls during flag evaluation for materialization reads/writes.

Custom Materialization Store

For advanced use cases, you can implement the MaterializationStore protocol to manage materialization data in your own infrastructure. The protocol defines two methods:

• read(ops: List[ReadOp]) -> List[ReadResult]: Batch read of materialization data
• write(ops: List[VariantWriteOp]) -> None: Batch write of variant assignments

The read operations support two types:

• VariantReadOp: Query for a sticky variant assignment (returns VariantReadResult)
• InclusionReadOp: Query for segment inclusion (returns InclusionReadResult)

from confidence.materialization import (
    MaterializationStore,
    ReadOp,
    ReadResult,
    VariantReadOp,
    VariantReadResult,
    InclusionReadOp,
    InclusionReadResult,
    VariantWriteOp,
)

class MyMaterializationStore:
    """Custom materialization store implementation."""

    def read(self, ops: list[ReadOp]) -> list[ReadResult]:
        results = []
        for op in ops:
            if isinstance(op, VariantReadOp):
                # Look up sticky variant assignment
                variant = self._lookup_variant(op.unit, op.materialization, op.rule)
                results.append(VariantReadResult(
                    unit=op.unit,
                    materialization=op.materialization,
                    rule=op.rule,
                    variant=variant,  # None if no assignment exists
                ))
            elif isinstance(op, InclusionReadOp):
                # Check segment inclusion
                included = self._check_inclusion(op.unit, op.materialization)
                results.append(InclusionReadResult(
                    unit=op.unit,
                    materialization=op.materialization,
                    included=included,
                ))
        return results

    def write(self, ops: list[VariantWriteOp]) -> None:
        for op in ops:
            # Store sticky variant assignment
            self._store_variant(op.unit, op.materialization, op.rule, op.variant)

Pass your custom store to the provider:

provider = ConfidenceProvider(
    client_secret="your-client-secret",
    materialization_store=MyMaterializationStore(),
)

Thread Safety: Your implementation must be thread-safe as it may be called concurrently from multiple threads.

Logging

Configure logging to see provider activity:

import logging
logging.getLogger("confidence").setLevel(logging.DEBUG)

Event Tracking

The provider supports the OpenFeature tracking API for sending custom events to the Confidence events backend. Event tracking is automatically enabled — no configuration needed.

📖 See the Integration Guide: Event Tracking for delivery guarantees, payload mapping rules, and cross-provider differences.

Usage

client = api.get_client()

context = EvaluationContext(
    targeting_key="user-123",
    attributes={"country": "US"},
)

# Track a simple event
client.track("checkout_completed", context)

# Track with a numeric value
client.track("purchase", context, TrackingEventDetails(value=49.99))

# Track with custom data
client.track("item_added", context, TrackingEventDetails(
    value=1,
    attributes={"sku": "ABC-123", "category": "electronics"},
))

Events are batched internally and flushed to the Confidence events service at the same interval as flag logs (configurable via log_poll_interval). On shutdown, pending events are drained on a best-effort basis (up to 100 batches).

Advanced: Controlling Exposure Events

By default, every flag evaluation records an exposure event (apply). Only disable this for exceptional cases where this provider must not collect exposures at all.

For normal feature delivery and experiments, keep applies enabled. When exposure collection is disabled, Confidence does not receive assignment/exposure events for those evaluations. Experiment results, exposure counts, assignment diagnostics, and downstream reporting that depend on exposures can be incomplete or unavailable. Resolve analytics and telemetry are still sent, so this is not a general logging or privacy-off switch.

To disable exposure collection for all OpenFeature evaluations through this provider, pass disable_exposure_collection=True when constructing the provider:

provider = ConfidenceProvider(
    client_secret="your-client-secret",
    disable_exposure_collection=True,
)

To skip exposure collection for a single evaluation, pass _confidence_skip_apply in the evaluation context:

context = EvaluationContext(
    targeting_key="user-123",
    attributes={"_confidence_skip_apply": True},
)

value = client.get_boolean_value("my-flag.enabled", False, context)

The key is automatically stripped from the context before it reaches the resolver.

This is an advanced feature intended for exceptional cases. If you're considering using it, reach out to the Confidence team to discuss the best approach for your setup.

License

Apache 2.0