feature-flags

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