Metadata-Version: 2.4
Name: flagkit
Version: 1.0.0
Summary: FlagKit SDK for Python - Feature flag evaluation and analytics
Project-URL: Homepage, https://flagkit.dev
Project-URL: Documentation, https://docs.flagkit.dev
Project-URL: Repository, https://github.com/teracrafts/flagkit-sdk
Project-URL: Issues, https://github.com/teracrafts/flagkit-sdk/issues
Author-email: FlagKit <sdk@flagkit.dev>
License-Expression: MIT
Keywords: feature-flags,feature-toggles,flagkit,sdk
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
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: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: cryptography>=41.0.0
Requires-Dist: httpx>=0.25.0
Provides-Extra: dev
Requires-Dist: mypy>=1.8.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest-cov>=4.1.0; extra == 'dev'
Requires-Dist: pytest-httpx>=0.28.0; extra == 'dev'
Requires-Dist: pytest>=7.4.0; extra == 'dev'
Requires-Dist: ruff>=0.1.9; extra == 'dev'
Description-Content-Type: text/markdown

# FlagKit Python SDK

Official Python SDK for [FlagKit](https://flagkit.dev) - Feature flag management made simple.

## Installation

```bash
pip install flagkit
```

## Requirements

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

## Quick Start

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

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

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

```python
from flagkit import EvaluationContext

# Set global context
context = EvaluationContext(
    user_id="user-123",
    email="user@example.com",
    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": "user@example.com"})

# 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

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

# Force flush pending events
client.flush()
```

### Lifecycle

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

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

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

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

```python
# 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": "user@example.com"})  # 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="user@example.com",
    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:

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

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

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

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

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

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

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

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

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

```bash
# 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](LICENSE) for details.
