flagkit 1.0.0

FlagKit SDK for Python - Feature flag evaluation and analytics

FlagKit Python SDK

Official Python SDK for FlagKit - Feature flag management made simple.

Installation

pip install flagkit

Requirements

• Python 3.9+
• httpx (automatically installed)
• cryptography (automatically installed, for cache encryption)

Quick Start

from flagkit import FlagKit

# Initialize the SDK
client = FlagKit.initialize(
    api_key="sdk_your_api_key",
    polling_interval=30,  # seconds
    cache_ttl=300,        # seconds
)

# Wait for initialization
client.wait_for_ready()

# Identify a user
client.identify("user-123", {"plan": "premium"})

# Evaluate flags
dark_mode = client.get_boolean_value("dark-mode", default=False)
welcome_message = client.get_string_value("welcome-message", default="Hello!")
max_items = client.get_number_value("max-items", default=10)
config = client.get_json_value("feature-config", default={})

# Get full evaluation details
result = client.evaluate("dark-mode")
print(f"Value: {result.value}, Reason: {result.reason}")

# Track custom events
client.track("button_clicked", {"button": "signup"})

# Cleanup when done
client.close()

Features

• Type-safe evaluation - Boolean, string, number, and JSON flag types
• Local caching - Fast evaluations with configurable TTL and optional encryption
• Background polling - Automatic flag updates with jitter
• Event tracking - Analytics with batching and crash-resilient persistence
• Resilient - Circuit breaker, retry with exponential backoff, offline support
• Thread-safe - Safe for concurrent use
• Security - PII detection, request signing, bootstrap verification, timing attack protection

Architecture

The SDK is organized into clean, modular packages:

flagkit/
├── __init__.py         # Public exports
├── flagkit.py          # Static FlagKit factory (singleton)
├── client.py           # FlagKitClient implementation
├── types/              # Public type definitions
│   ├── config.py       # FlagKitOptions, BootstrapConfig
│   ├── context.py      # EvaluationContext
│   ├── flag.py         # FlagState, EvaluationResult
│   └── events.py       # Event types
├── errors/             # Error types and codes
│   ├── error_codes.py  # 31 error codes
│   ├── flagkit_error.py # Error hierarchy
│   └── sanitizer.py    # Error message sanitization
├── http/               # HTTP client, circuit breaker, retry
│   ├── http_client.py  # HTTP with retry/circuit integration
│   ├── circuit_breaker.py
│   └── retry.py        # Exponential backoff with jitter
├── core/               # Core components
│   ├── cache.py        # In-memory cache with TTL
│   ├── context_manager.py
│   ├── polling_manager.py # Background polling
│   ├── event_queue.py  # Event batching
│   └── event_persistence.py # Crash-resilient persistence
├── storage/            # Storage implementations
│   ├── storage.py      # Storage Protocol
│   ├── memory_storage.py
│   └── encrypted_storage.py # AES-256-GCM encryption
└── utils/              # Utilities
    ├── logger.py       # Logger interface
    ├── security.py     # PII detection, HMAC signing
    ├── platform.py     # Platform detection
    └── validators.py   # Input validation

API Reference

Initialization

from flagkit import FlagKit, FlagKitClient, FlagKitOptions

# Using the static factory (recommended for single client)
client = FlagKit.initialize(
    api_key="sdk_...",              # Required
    base_url="https://api.flagkit.dev/api/v1",  # Optional
    polling_interval=30.0,          # Seconds between polls
    enable_polling=True,            # Enable background polling
    cache_enabled=True,             # Enable local caching
    cache_ttl=300.0,                # Cache TTL in seconds
    offline=False,                  # Offline mode
    timeout=5.0,                    # Request timeout in seconds
    retries=3,                      # Number of retries
    bootstrap={"flag": True},       # Initial flag values
    debug=False,                    # Enable debug logging
    on_ready=lambda: print("Ready!"),
    on_error=lambda e: print(f"Error: {e}"),
    on_update=lambda flags: print(f"Updated: {len(flags)} flags"),
)

# Or create a client directly
options = FlagKitOptions(api_key="sdk_...")
client = FlagKitClient(options)
client.initialize()

Flag Evaluation

# Boolean flags
enabled = client.get_boolean_value("feature-flag", default=False)

# String flags
variant = client.get_string_value("button-text", default="Click")

# Number flags
limit = client.get_number_value("rate-limit", default=100)

# JSON flags
config = client.get_json_value("config", default={"enabled": False})

# Full evaluation result
result = client.evaluate("feature-flag")
# result.flag_key, result.value, result.enabled, result.reason, result.version

# Evaluate all flags
all_results = client.evaluate_all()

# Check flag existence
if client.has_flag("my-flag"):
    # ...

# Get all flag keys
keys = client.get_all_flag_keys()

Context Management

from flagkit import EvaluationContext

# Set global context
context = EvaluationContext(
    user_id="user-123",
    email="[e-mail]",
    country="US",
    custom={"plan": "premium", "beta_tester": True},
    private_attributes=["email"],  # Not sent to server
)
client.set_context(context)

# Get current context
current = client.get_context()

# Clear context
client.clear_context()

# Identify user (shorthand)
client.identify("user-123", {"email": "[e-mail]"})

# Reset to anonymous
client.reset()

# Pass context to evaluation
result = client.get_boolean_value(
    "feature-flag",
    default=False,
    context=EvaluationContext(user_id="other-user"),
)

Event Tracking

# Track custom event
client.track("purchase", {
    "amount": 99.99,
    "currency": "USD",
    "product_id": "prod-123",
})

# Force flush pending events
client.flush()

Lifecycle

# Check if SDK is ready
if client.is_ready():
    # ...

# Wait for ready (blocks until initialized)
client.wait_for_ready()

# Force refresh flags from server
client.refresh()

# Close SDK and cleanup
client.close()

# Using context manager
with FlagKitClient(options) as client:
    client.initialize()
    # Use client...
# Automatically closed

Error Handling

from flagkit import FlagKitError, InitializationError, NetworkError

try:
    client = FlagKit.initialize(api_key="sdk_...")
except InitializationError as e:
    print(f"Failed to initialize: {e.code} - {e}")
except NetworkError as e:
    if e.recoverable:
        # Retry logic
        pass
except FlagKitError as e:
    print(f"Error [{e.code}]: {e}")
    print(f"Recoverable: {e.recoverable}")
    print(f"Details: {e.to_dict()}")

Local Development

# Connect to local FlagKit server at http://localhost:8200/api/v1
client = FlagKit.initialize(
    api_key="sdk_...",
    local_port=8200,
)

# Or use a custom port
client = FlagKit.initialize(
    api_key="sdk_...",
    local_port=3000,  # Uses http://localhost:3000/api/v1
)

Offline Mode

# Start in offline mode
client = FlagKit.initialize(
    api_key="sdk_...",
    offline=True,
    bootstrap={"feature-flag": True},
)

# Uses bootstrap values without network requests
value = client.get_boolean_value("feature-flag", default=False)

Security Features

PII Detection

The SDK can detect and warn about potential PII (Personally Identifiable Information) in contexts and events:

# Enable strict PII mode - raises errors instead of warnings
client = FlagKit.initialize(
    api_key="sdk_...",
    strict_pii_mode=True,
)

# Attributes containing PII will raise SecurityError
try:
    client.identify("user-123", {"email": "[e-mail]"})  # PII detected!
except SecurityError as e:
    print(f"PII error: {e}")

# Use private attributes to mark fields as intentionally containing PII
context = EvaluationContext(
    user_id="user-123",
    email="[e-mail]",
    private_attributes=["email"],  # Marks email as intentionally private
)
client.set_context(context)  # No error - email marked as private

Request Signing

POST requests to the FlagKit API are signed with HMAC-SHA256 for integrity:

# Enabled by default, can be disabled if needed
client = FlagKit.initialize(
    api_key="sdk_...",
    enable_request_signing=False,  # Disable signing
)

Bootstrap Signature Verification

Verify bootstrap data integrity using HMAC signatures:

from flagkit import BootstrapConfig, BootstrapVerificationOptions
from flagkit.utils.security import create_bootstrap_signature

# Create signed bootstrap data
bootstrap = create_bootstrap_signature(
    flags={"feature-a": True, "feature-b": "value"},
    api_key="sdk_your_api_key",
)

# Use signed bootstrap with verification
client = FlagKit.initialize(
    api_key="sdk_...",
    bootstrap=bootstrap,
    bootstrap_verification=BootstrapVerificationOptions(
        enabled=True,
        max_age=86400000,  # 24 hours in milliseconds
        on_failure="error",  # "warn" (default), "error", or "ignore"
    ),
)

Cache Encryption

Enable AES-256-GCM encryption for cached flag data:

client = FlagKit.initialize(
    api_key="sdk_...",
    encrypt_cache=True,
)

Evaluation Jitter (Timing Attack Protection)

Add random delays to flag evaluations to prevent cache timing attacks:

from flagkit import EvaluationJitterConfig

client = FlagKit.initialize(
    api_key="sdk_...",
    evaluation_jitter=EvaluationJitterConfig(
        enabled=True,
        min_ms=5,
        max_ms=15,
    ),
)

Error Sanitization

Automatically redact sensitive information from error messages:

from flagkit import ErrorSanitizationConfig

client = FlagKit.initialize(
    api_key="sdk_...",
    error_sanitization=ErrorSanitizationConfig(
        enabled=True,
        preserve_original=False,  # Set True to keep original for debugging
    ),
)
# Errors will have paths, IPs, API keys, and emails redacted

Event Persistence

Enable crash-resilient event persistence to prevent data loss:

client = FlagKit.initialize(
    api_key="sdk_...",
    persist_events=True,
    event_storage_path="/path/to/storage",  # Optional, defaults to temp dir
    max_persisted_events=10000,             # Optional, default 10000
    persistence_flush_interval=1000,        # Optional, default 1000ms
)

Events are written to disk before being sent, and automatically recovered on restart.

Key Rotation

Support seamless API key rotation:

client = FlagKit.initialize(
    api_key="sdk_primary_key",
    secondary_api_key="sdk_secondary_key",
    key_rotation_grace_period=300.0,  # 5 minutes
)
# SDK will automatically failover to secondary key on 401 errors

Error Handling

from flagkit import (
    FlagKitError,
    InitializationError,
    NetworkError,
    SecurityError,
)

try:
    client = FlagKit.initialize(api_key="sdk_...")
except InitializationError as e:
    print(f"Failed to initialize: {e.code} - {e}")
except SecurityError as e:
    print(f"Security error: {e.code} - {e}")
except NetworkError as e:
    if e.recoverable:
        # Retry logic
        pass
except FlagKitError as e:
    print(f"Error [{e.code}]: {e}")
    print(f"Recoverable: {e.recoverable}")
    print(f"Details: {e.to_dict()}")

Error Codes

Code	Description
INIT_FAILED	SDK initialization failed
INIT_TIMEOUT	Initialization timed out
INIT_ALREADY_INITIALIZED	SDK already initialized
INIT_NOT_INITIALIZED	SDK not initialized
NETWORK_ERROR	Network request failed
AUTH_INVALID_KEY	Invalid API key
SECURITY_PII_DETECTED	PII detected in strict mode
SECURITY_LOCAL_PORT_IN_PRODUCTION	Local port used in production
SECURITY_SIGNATURE_INVALID	Bootstrap signature verification failed

Configuration Options

Option	Type	Default	Description
api_key	str	Required	API key for authentication
secondary_api_key	str	None	Secondary key for rotation
key_rotation_grace_period	float	300.0	Grace period in seconds
base_url	str	https://api.flagkit.dev/api/v1	API base URL
local_port	int	None	Local development port
polling_interval	float	30.0	Polling interval in seconds
enable_polling	bool	True	Enable background polling
cache_enabled	bool	True	Enable local caching
cache_ttl	float	300.0	Cache TTL in seconds
encrypt_cache	bool	False	Enable AES-256-GCM cache encryption
offline	bool	False	Offline mode
timeout	float	5.0	Request timeout in seconds
retries	int	3	Number of retry attempts
bootstrap	dict/BootstrapConfig	{}	Initial flag values
bootstrap_verification	BootstrapVerificationOptions	enabled	Bootstrap verification settings
debug	bool	False	Enable debug logging
logger	LoggerProtocol	None	Custom logger
on_ready	Callable	None	Ready callback
on_error	Callable	None	Error callback
on_update	Callable	None	Update callback
enable_request_signing	bool	True	Enable request signing
strict_pii_mode	bool	False	Error on PII detection
persist_events	bool	False	Enable event persistence
event_storage_path	str	temp dir	Event storage directory
max_persisted_events	int	10000	Max persisted events
persistence_flush_interval	int	1000	Persistence flush interval (ms)
evaluation_jitter	EvaluationJitterConfig	disabled	Timing attack protection
error_sanitization	ErrorSanitizationConfig	enabled	Redact sensitive info from errors

Testing

# Use offline mode with bootstrap values
client = FlagKit.initialize(
    api_key="sdk_test",
    offline=True,
    bootstrap={"feature-flag": True},
)
client.wait_for_ready()

# With signed bootstrap for verification testing
from flagkit.utils.security import create_bootstrap_signature

bootstrap = create_bootstrap_signature(
    flags={"feature-flag": True},
    api_key="sdk_test",
)
client = FlagKit.initialize(
    api_key="sdk_test",
    offline=True,
    bootstrap=bootstrap,
)

Thread Safety

All SDK methods are safe for concurrent use from multiple threads. The client uses internal synchronization (threading.Lock) to ensure thread-safe access to:

• Flag cache
• Event queue
• Context management
• Polling state

Development

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run with coverage
pytest --cov

# Type checking
mypy src/

# Linting
ruff check src/

License

MIT License - see LICENSE for details.