ffsshhttiikk What I do
- Implement feature flags
- Manage feature rollouts
- Implement A/B testing
- Handle percentage rollouts
- Target specific users
- Implement flag dependencies
- Monitor flag performance
- Clean up deprecated flags
When to use me
When implementing feature flags or managing feature rollouts.
Feature Flag Architecture
┌─────────────────────────────────────────────────────────────┐
│ Feature Flag System │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Toggle │ │ Router │ │ Evaluator │ │
│ │ Engine │ │ Engine │ │ Engine │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ └───────────────────┼───────────────────┘ │
│ │ │
│ ┌───────────────────────────▼───────────────────────────┐ │
│ │ Feature Flag Storage │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────────────┐ │ │
│ │ │ Redis │ │Database │ │ Config Files │ │ │
│ │ └─────────┘ └─────────┘ └─────────────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
Feature Flag Implementation
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional
from datetime import datetime
import json
@dataclass
class FeatureFlag:
"""Feature flag configuration."""
key: str
name: str
description: str = ""
enabled: bool = False
percentage: int = 0 # 0-100
target_users: list = None
exclude_users: list = None
strategies: list = None
dependencies: list = None
created_at: datetime = None
updated_at: datetime = None
metadata: Dict[str, Any] = None
def __post_init__(self):
self.target_users = self.target_users or []
self.exclude_users = self.exclude_users or []
self.strategies = self.strategies or []
self.dependencies = self.dependencies or []
self.metadata = self.metadata or {}
class FeatureFlagEngine:
"""Feature flag evaluation engine."""
def __init__(self, storage):
self.storage = storage
self.cache = {}
self.cache_ttl = 60 # seconds
def is_enabled(
self,
flag_key: str,
user_id: str = None,
context: Dict[str, Any] = None
) -> bool:
"""Check if a feature flag is enabled."""
flag = self._get_flag(flag_key)
if not flag:
return False
if not flag.enabled:
return False
# Check dependencies first
for dep_key in flag.dependencies:
if not self.is_enabled(dep_key, user_id, context):
return False
# Check if user is in target list
if flag.target_users and user_id in flag.target_users:
return True
# Check if user is excluded
if flag.exclude_users and user_id in flag.exclude_users:
return False
# Check percentage rollout
if flag.percentage > 0:
return self._in_percentage(flag_key, user_id, flag.percentage)
# Check custom strategies
for strategy in flag.strategies:
if self._evaluate_strategy(strategy, user_id, context):
return True
return flag.enabled
def _get_flag(self, flag_key: str) -> Optional[FeatureFlag]:
"""Get flag from cache or storage."""
if flag_key in self.cache:
cached_flag, timestamp = self.cache[flag_key]
if (datetime.utcnow() - timestamp).seconds < self.cache_ttl:
return cached_flag
flag = self.storage.get_flag(flag_key)
if flag:
self.cache[flag_key] = (flag, datetime.utcnow())
return flag
def _in_percentage(
self,
flag_key: str,
user_id: str,
percentage: int
) -> bool:
"""Determine if user is in percentage rollout."""
if not user_id:
user_id = "anonymous"
# Consistent hashing based on user_id
hash_value = hash(f"{flag_key}:{user_id}")
bucket = hash_value % 100
return bucket < percentage
def _evaluate_strategy(
self,
strategy: Dict[str, Any],
user_id: str,
context: Dict[str, Any]
) -> bool:
"""Evaluate custom strategy."""
strategy_type = strategy.get("type")
if strategy_type == "user_id":
return user_id in strategy.get("values", [])
if strategy_type == "attribute":
attr = strategy.get("attribute")
op = strategy.get("operator")
value = strategy.get("value")
user_attr = context.get(attr) if context else None
return self._compare(user_attr, op, value)
return False
def _compare(self, actual, operator: str, expected) -> bool:
"""Compare values based on operator."""
operators = {
"equals": lambda a, e: a == e,
"not_equals": lambda a, e: a != e,
"contains": lambda a, e: e in a if a else False,
"starts_with": lambda a, e: a.startswith(e) if a else False,
"ends_with": lambda a, e: a.endswith(e) if a else False,
"greater_than": lambda a, e: a > e if a and e else False,
"less_than": lambda a, e: a < e if a and e else False,
}
op_func = operators.get(operator)
if op_func:
return op_func(actual, expected)
return False
# Decorator for feature flags
def feature_flag(flag_key: str, default: bool = False):
"""Decorator to conditionally enable functionality."""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
engine = get_feature_flag_engine()
# Get user from request context
user_id = get_current_user_id()
if engine.is_enabled(flag_key, user_id):
return func(*args, **kwargs)
else:
return default
return wrapper
return decorator
# Usage
@feature_flag("new-checkout-flow", default=None)
async def process_checkout():
"""New checkout flow."""
return await process_new_checkout()
@feature_flag("beta-analytics")
def get_analytics_dashboard():
"""Get beta analytics dashboard."""
return render_beta_dashboard()
# Context manager for features
class FeatureContext:
"""Context for feature flag evaluation."""
def __init__(self, user_id: str = None, **attributes):
self.user_id = user_id
self.attributes = attributes
def is_enabled(self, flag_key: str) -> bool:
engine = get_feature_flag_engine()
return engine.is_enabled(flag_key, self.user_id, self.attributes)
# Usage
context = FeatureContext(
user_id="user_123",
plan="enterprise",
region="us-east",
age_days=30
)
if context.is_enabled("new-feature"):
show_new_feature()
else:
show_legacy_feature()
A/B Testing with Flags
@dataclass
class ABTest:
"""A/B test configuration."""
name: str
variants: list[str]
weights: list[int]
metric_name: str
description: str = ""
class ABTestEngine:
"""A/B test evaluation engine."""
def __init__(self, flag_engine: FeatureFlagEngine):
self.flag_engine = flag_engine
self.tests: Dict[str, ABTest] = {}
self.user_assignments: Dict[str, str] = {}
def register_test(self, test: ABTest) -> None:
"""Register a new A/B test."""
self.tests[test.name] = test
def get_variant(
self,
test_name: str,
user_id: str = None
) -> Optional[str]:
"""Get which variant a user is in."""
if test_name not in self.tests:
return None
assignment_key = f"{test_name}:{user_id}"
# Check existing assignment
if assignment_key in self.user_assignments:
return self.user_assignments[assignment_key]
test = self.tests[test_name]
# Assign variant
variant = self._assign_variant(test_name, test, user_id)
self.user_assignments[assignment_key] = variant
return variant
def _assign_variant(
self,
test_name: str,
test: ABTest,
user_id: str
) -> str:
"""Assign user to variant based on weights."""
if not user_id:
user_id = "anonymous"
# Hash for consistent assignment
hash_value = hash(f"{test_name}:{user_id}")
bucket = hash_value % 100
# Assign based on cumulative weights
cumulative = 0
for variant, weight in zip(test.variants, test.weights):
cumulative += weight
if bucket < cumulative:
return variant
return test.variants[-1] # Default to last variant
def track_event(
self,
test_name: str,
user_id: str,
event_name: str,
properties: Dict[str, Any] = None
) -> None:
"""Track A/B test event."""
variant = self.get_variant(test_name, user_id)
if variant:
# Send to analytics
analytics.track(
event=f"ab_test:{event_name}",
user_id=user_id,
properties={
"test_name": test_name,
"variant": variant,
**(properties or {})
}
)
def get_test_results(self, test_name: str) -> Dict[str, Any]:
"""Get A/B test results."""
# Aggregate events by variant
# Calculate conversion rates
# Return statistical analysis
pass
Best Practices
Feature Flag Best Practices:
1. Keep flags simple
One flag per feature
Avoid complex conditions
2. Use meaningful names
Descriptive flag names
Include feature context
3. Document flags
Purpose of each flag
Expected removal date
4. Remove old flags
Technical debt
Clean up after rollout
5. Use proper targeting
Percentage rollouts
User segmentation
6. Monitor performance
Track flag metrics
Performance impact
7. Test in production
Canary releases
Gradual rollouts
8. Separate concerns
Feature flags from config
Don't use for environment
9. Version control
Store flags in repo
Review flag changes
10. Have a process
Approval for production
Rollback procedures