#!/usr/bin/env python3 """ GENESIS FEATURE FLAGS SYSTEM ============================== Dynamic feature flag management with targeting and rollouts. Features: - Boolean, string, number, JSON flag types - User/group targeting - Percentage rollouts - A/B testing support - Environment-based flags - Real-time flag updates - Audit logging Usage: flags = FeatureFlags() flags.define("new_feature", default=False) if flags.is_enabled("new_feature", user_id="123"): # New feature code else: # Old feature code @feature_flag("beta_feature") def new_function(): return "new implementation" """ import hashlib import json import random import threading import time from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from enum import Enum from functools import wraps from pathlib import Path from typing import Dict, List, Any, Optional, Callable, Union, Set class FlagType(Enum): """Types of feature flags.""" BOOLEAN = "boolean" STRING = "string" NUMBER = "number" JSON = "json" class RolloutStrategy(Enum): """Rollout strategies for feature flags.""" ALL = "all" # All users NONE = "none" # No users PERCENTAGE = "percentage" # Percentage of users USER_IDS = "user_ids" # Specific user IDs GROUPS = "groups" # User groups GRADUAL = "gradual" # Gradual rollout over time RANDOM = "random" # Random selection each time @dataclass class TargetingRule: """Rule for targeting specific users/groups.""" attribute: str # User attribute to match operator: str # eq, ne, contains, gt, lt, in, not_in value: Any # Value to compare against enabled: bool = True # Whether this rule enables or disables @dataclass class FlagVariant: """A variant for A/B testing.""" name: str value: Any weight: int = 1 # Relative weight for selection @dataclass class FlagDefinition: """Definition of a feature flag.""" name: str flag_type: FlagType = FlagType.BOOLEAN default_value: Any = False description: str = "" # Rollout configuration strategy: RolloutStrategy = RolloutStrategy.ALL percentage: float = 100.0 # For percentage rollout target_users: Set[str] = field(default_factory=set) target_groups: Set[str] = field(default_factory=set) # Targeting rules rules: List[TargetingRule] = field(default_factory=list) # A/B variants variants: List[FlagVariant] = field(default_factory=list) # Environment environments: Set[str] = field(default_factory=set) # Empty = all # Metadata enabled: bool = True created_at: str = "" updated_at: str = "" tags: List[str] = field(default_factory=list) def to_dict(self) -> Dict: return { "name": self.name, "type": self.flag_type.value, "default": self.default_value, "description": self.description, "strategy": self.strategy.value, "percentage": self.percentage, "enabled": self.enabled, "tags": self.tags } @dataclass class EvaluationContext: """Context for flag evaluation.""" user_id: Optional[str] = None groups: Set[str] = field(default_factory=set) attributes: Dict[str, Any] = field(default_factory=dict) environment: str = "production" @dataclass class EvaluationResult: """Result of flag evaluation.""" flag_name: str enabled: bool value: Any variant: Optional[str] = None reason: str = "" @dataclass class AuditEntry: """Audit log entry for flag changes.""" timestamp: str flag_name: str action: str # create, update, delete, evaluate old_value: Any = None new_value: Any = None user_id: Optional[str] = None class FlagBackend(ABC): """Abstract backend for flag storage.""" @abstractmethod def get(self, name: str) -> Optional[FlagDefinition]: pass @abstractmethod def set(self, flag: FlagDefinition) -> bool: pass @abstractmethod def delete(self, name: str) -> bool: pass @abstractmethod def list(self) -> List[str]: pass class MemoryBackend(FlagBackend): """In-memory flag storage.""" def __init__(self): self._flags: Dict[str, FlagDefinition] = {} self._lock = threading.RLock() def get(self, name: str) -> Optional[FlagDefinition]: with self._lock: return self._flags.get(name) def set(self, flag: FlagDefinition) -> bool: with self._lock: self._flags[flag.name] = flag return True def delete(self, name: str) -> bool: with self._lock: if name in self._flags: del self._flags[name] return True return False def list(self) -> List[str]: with self._lock: return list(self._flags.keys()) class FileBackend(FlagBackend): """File-based flag storage.""" def __init__(self, path: Path): self.path = path self.path.parent.mkdir(parents=True, exist_ok=True) self._flags: Dict[str, FlagDefinition] = {} self._lock = threading.RLock() self._load() def _load(self): if self.path.exists(): try: with open(self.path, 'r') as f: data = json.load(f) for name, flag_data in data.get("flags", {}).items(): self._flags[name] = self._dict_to_flag(flag_data) except Exception: pass def _save(self): data = { "flags": { name: self._flag_to_dict(flag) for name, flag in self._flags.items() } } with open(self.path, 'w') as f: json.dump(data, f, indent=2) def _flag_to_dict(self, flag: FlagDefinition) -> Dict: return { "name": flag.name, "type": flag.flag_type.value, "default": flag.default_value, "description": flag.description, "strategy": flag.strategy.value, "percentage": flag.percentage, "target_users": list(flag.target_users), "target_groups": list(flag.target_groups), "enabled": flag.enabled, "tags": flag.tags, "created_at": flag.created_at, "updated_at": flag.updated_at } def _dict_to_flag(self, data: Dict) -> FlagDefinition: return FlagDefinition( name=data["name"], flag_type=FlagType(data.get("type", "boolean")), default_value=data.get("default", False), description=data.get("description", ""), strategy=RolloutStrategy(data.get("strategy", "all")), percentage=data.get("percentage", 100.0), target_users=set(data.get("target_users", [])), target_groups=set(data.get("target_groups", [])), enabled=data.get("enabled", True), tags=data.get("tags", []), created_at=data.get("created_at", ""), updated_at=data.get("updated_at", "") ) def get(self, name: str) -> Optional[FlagDefinition]: with self._lock: return self._flags.get(name) def set(self, flag: FlagDefinition) -> bool: with self._lock: self._flags[flag.name] = flag self._save() return True def delete(self, name: str) -> bool: with self._lock: if name in self._flags: del self._flags[name] self._save() return True return False def list(self) -> List[str]: with self._lock: return list(self._flags.keys()) class FeatureFlags: """ Central feature flags management system. """ def __init__( self, backend: FlagBackend = None, environment: str = "production", enable_audit: bool = True ): self.backend = backend or MemoryBackend() self.environment = environment self._enable_audit = enable_audit self._audit_log: List[AuditEntry] = [] self._lock = threading.RLock() # Callbacks self._on_change: List[Callable[[str, Any, Any], None]] = [] def define( self, name: str, default: Any = False, flag_type: FlagType = None, description: str = "", strategy: RolloutStrategy = RolloutStrategy.ALL, percentage: float = 100.0, target_users: List[str] = None, target_groups: List[str] = None, environments: List[str] = None, tags: List[str] = None ) -> FlagDefinition: """Define a new feature flag.""" # Infer type if not provided if flag_type is None: if isinstance(default, bool): flag_type = FlagType.BOOLEAN elif isinstance(default, str): flag_type = FlagType.STRING elif isinstance(default, (int, float)): flag_type = FlagType.NUMBER else: flag_type = FlagType.JSON now = datetime.now().isoformat() flag = FlagDefinition( name=name, flag_type=flag_type, default_value=default, description=description, strategy=strategy, percentage=percentage, target_users=set(target_users or []), target_groups=set(target_groups or []), environments=set(environments or []), tags=tags or [], enabled=True, created_at=now, updated_at=now ) self.backend.set(flag) self._audit("create", name, new_value=flag.to_dict()) return flag def evaluate( self, name: str, context: EvaluationContext = None ) -> EvaluationResult: """Evaluate a feature flag for the given context.""" flag = self.backend.get(name) if not flag: return EvaluationResult( flag_name=name, enabled=False, value=None, reason="flag_not_found" ) context = context or EvaluationContext() # Check if flag is globally disabled if not flag.enabled: return EvaluationResult( flag_name=name, enabled=False, value=flag.default_value, reason="flag_disabled" ) # Check environment if flag.environments and self.environment not in flag.environments: return EvaluationResult( flag_name=name, enabled=False, value=flag.default_value, reason="environment_mismatch" ) # Apply strategy enabled = self._apply_strategy(flag, context) # Apply targeting rules if flag.rules: for rule in flag.rules: if self._evaluate_rule(rule, context): enabled = rule.enabled break # Select variant if A/B testing variant = None value = flag.default_value if enabled and flag.variants: variant = self._select_variant(flag, context) if variant: value = variant.value self._audit("evaluate", name, user_id=context.user_id) return EvaluationResult( flag_name=name, enabled=enabled, value=value if enabled else flag.default_value, variant=variant.name if variant else None, reason="evaluated" ) def is_enabled( self, name: str, user_id: str = None, groups: List[str] = None, attributes: Dict = None ) -> bool: """Check if a feature flag is enabled (convenience method).""" context = EvaluationContext( user_id=user_id, groups=set(groups or []), attributes=attributes or {}, environment=self.environment ) return self.evaluate(name, context).enabled def get_value( self, name: str, default: Any = None, user_id: str = None, **kwargs ) -> Any: """Get the value of a feature flag.""" context = EvaluationContext( user_id=user_id, environment=self.environment, **kwargs ) result = self.evaluate(name, context) return result.value if result.enabled else default def _apply_strategy( self, flag: FlagDefinition, context: EvaluationContext ) -> bool: """Apply rollout strategy.""" if flag.strategy == RolloutStrategy.ALL: return True elif flag.strategy == RolloutStrategy.NONE: return False elif flag.strategy == RolloutStrategy.PERCENTAGE: if context.user_id: # Consistent hashing for same user hash_value = int( hashlib.md5( f"{flag.name}:{context.user_id}".encode() ).hexdigest(), 16 ) return (hash_value % 100) < flag.percentage else: return random.random() * 100 < flag.percentage elif flag.strategy == RolloutStrategy.USER_IDS: return context.user_id in flag.target_users elif flag.strategy == RolloutStrategy.GROUPS: return bool(context.groups & flag.target_groups) elif flag.strategy == RolloutStrategy.RANDOM: return random.random() * 100 < flag.percentage return True def _evaluate_rule( self, rule: TargetingRule, context: EvaluationContext ) -> bool: """Evaluate a targeting rule.""" attr_value = context.attributes.get(rule.attribute) if attr_value is None: return False op = rule.operator target = rule.value if op == "eq": return attr_value == target elif op == "ne": return attr_value != target elif op == "contains": return target in str(attr_value) elif op == "gt": return attr_value > target elif op == "lt": return attr_value < target elif op == "gte": return attr_value >= target elif op == "lte": return attr_value <= target elif op == "in": return attr_value in target elif op == "not_in": return attr_value not in target elif op == "regex": import re return bool(re.match(target, str(attr_value))) return False def _select_variant( self, flag: FlagDefinition, context: EvaluationContext ) -> Optional[FlagVariant]: """Select a variant for A/B testing.""" if not flag.variants: return None # Calculate total weight total_weight = sum(v.weight for v in flag.variants) if total_weight == 0: return flag.variants[0] # Consistent selection based on user if context.user_id: hash_value = int( hashlib.md5( f"{flag.name}:variant:{context.user_id}".encode() ).hexdigest(), 16 ) selection = hash_value % total_weight else: selection = random.randint(0, total_weight - 1) # Select variant cumulative = 0 for variant in flag.variants: cumulative += variant.weight if selection < cumulative: return variant return flag.variants[-1] def update( self, name: str, enabled: bool = None, percentage: float = None, target_users: List[str] = None, target_groups: List[str] = None, default_value: Any = None ) -> bool: """Update a feature flag.""" flag = self.backend.get(name) if not flag: return False old_value = flag.to_dict() if enabled is not None: flag.enabled = enabled if percentage is not None: flag.percentage = percentage if target_users is not None: flag.target_users = set(target_users) if target_groups is not None: flag.target_groups = set(target_groups) if default_value is not None: flag.default_value = default_value flag.updated_at = datetime.now().isoformat() self.backend.set(flag) self._audit("update", name, old_value=old_value, new_value=flag.to_dict()) # Notify callbacks for callback in self._on_change: try: callback(name, old_value, flag.to_dict()) except Exception: pass return True def delete(self, name: str) -> bool: """Delete a feature flag.""" flag = self.backend.get(name) if flag: self._audit("delete", name, old_value=flag.to_dict()) return self.backend.delete(name) def list_flags(self, tag: str = None) -> List[FlagDefinition]: """List all flags, optionally filtered by tag.""" flags = [] for name in self.backend.list(): flag = self.backend.get(name) if flag: if tag is None or tag in flag.tags: flags.append(flag) return flags def get_flag(self, name: str) -> Optional[FlagDefinition]: """Get a flag definition.""" return self.backend.get(name) def on_change(self, callback: Callable[[str, Any, Any], None]): """Register callback for flag changes.""" self._on_change.append(callback) def _audit( self, action: str, flag_name: str, old_value: Any = None, new_value: Any = None, user_id: str = None ): """Log an audit entry.""" if not self._enable_audit: return entry = AuditEntry( timestamp=datetime.now().isoformat(), flag_name=flag_name, action=action, old_value=old_value, new_value=new_value, user_id=user_id ) self._audit_log.append(entry) if len(self._audit_log) > 1000: self._audit_log = self._audit_log[-1000:] def get_audit_log(self, limit: int = 100, flag_name: str = None) -> List[AuditEntry]: """Get audit log entries.""" log = self._audit_log.copy() if flag_name: log = [e for e in log if e.flag_name == flag_name] return log[-limit:] def get_status(self) -> Dict: """Get feature flags status.""" flags = self.list_flags() enabled_count = sum(1 for f in flags if f.enabled) return { "total_flags": len(flags), "enabled_flags": enabled_count, "disabled_flags": len(flags) - enabled_count, "environment": self.environment, "audit_entries": len(self._audit_log) } # Decorator for feature-flagged functions def feature_flag( flag_name: str, fallback: Callable = None, flags: FeatureFlags = None ): """Decorator to wrap a function with a feature flag.""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): ff = flags or _global_flags if ff.is_enabled(flag_name): return func(*args, **kwargs) elif fallback: return fallback(*args, **kwargs) else: return None @wraps(func) async def async_wrapper(*args, **kwargs): ff = flags or _global_flags if ff.is_enabled(flag_name): return await func(*args, **kwargs) elif fallback: if asyncio.iscoroutinefunction(fallback): return await fallback(*args, **kwargs) return fallback(*args, **kwargs) else: return None import asyncio if asyncio.iscoroutinefunction(func): return async_wrapper return wrapper return decorator # Global feature flags instance _global_flags: Optional[FeatureFlags] = None def get_flags() -> FeatureFlags: """Get global feature flags instance.""" global _global_flags if _global_flags is None: _global_flags = FeatureFlags() return _global_flags def main(): """CLI and demo for feature flags.""" import argparse parser = argparse.ArgumentParser(description="Genesis Feature Flags") parser.add_argument("command", choices=["demo", "list", "status"]) args = parser.parse_args() if args.command == "demo": print("Feature Flags Demo") print("=" * 40) ff = FeatureFlags() # Basic boolean flag print("\n1. Basic Boolean Flag:") ff.define("new_dashboard", default=False, description="New dashboard UI") ff.update("new_dashboard", enabled=True) print(f" new_dashboard enabled: {ff.is_enabled('new_dashboard')}") # Percentage rollout print("\n2. Percentage Rollout (50%):") ff.define( "beta_feature", default=False, strategy=RolloutStrategy.PERCENTAGE, percentage=50.0 ) enabled_count = sum( 1 for i in range(100) if ff.is_enabled("beta_feature", user_id=f"user_{i}") ) print(f" Enabled for {enabled_count}/100 users") # User targeting print("\n3. User Targeting:") ff.define( "admin_feature", default=False, strategy=RolloutStrategy.USER_IDS, target_users=["admin_1", "admin_2"] ) print(f" admin_1: {ff.is_enabled('admin_feature', user_id='admin_1')}") print(f" regular_user: {ff.is_enabled('admin_feature', user_id='regular_user')}") # Group targeting print("\n4. Group Targeting:") ff.define( "enterprise_feature", default=False, strategy=RolloutStrategy.GROUPS, target_groups=["enterprise", "premium"] ) print(f" enterprise group: {ff.is_enabled('enterprise_feature', groups=['enterprise'])}") print(f" free group: {ff.is_enabled('enterprise_feature', groups=['free'])}") # A/B Testing print("\n5. A/B Testing:") ff.define("checkout_flow", default="control") flag = ff.get_flag("checkout_flow") flag.variants = [ FlagVariant(name="control", value="original", weight=50), FlagVariant(name="variant_a", value="simplified", weight=30), FlagVariant(name="variant_b", value="express", weight=20) ] ff.backend.set(flag) variant_counts: Dict[str, int] = {} for i in range(100): result = ff.evaluate( "checkout_flow", EvaluationContext(user_id=f"user_{i}") ) variant = result.variant or "control" variant_counts[variant] = variant_counts.get(variant, 0) + 1 print(f" Variant distribution: {json.dumps(variant_counts, indent=4)}") # String flag print("\n6. String Flag:") ff.define("api_version", default="v1", flag_type=FlagType.STRING) print(f" API version: {ff.get_value('api_version')}") # Feature flag decorator print("\n7. Feature Flag Decorator:") @feature_flag("new_algorithm", fallback=lambda: "old result", flags=ff) def compute(): return "new result" ff.define("new_algorithm", default=False) print(f" With flag disabled: {compute()}") ff.update("new_algorithm", enabled=True) print(f" With flag enabled: {compute()}") # Targeting rules print("\n8. Targeting Rules:") ff.define("premium_content", default=False) flag = ff.get_flag("premium_content") flag.rules = [ TargetingRule( attribute="subscription", operator="in", value=["premium", "enterprise"], enabled=True ) ] ff.backend.set(flag) ctx1 = EvaluationContext(attributes={"subscription": "premium"}) ctx2 = EvaluationContext(attributes={"subscription": "free"}) print(f" Premium user: {ff.evaluate('premium_content', ctx1).enabled}") print(f" Free user: {ff.evaluate('premium_content', ctx2).enabled}") # Status print("\n9. Status:") print(f" {json.dumps(ff.get_status(), indent=4)}") # Audit log print("\n10. Audit Log (last 5):") for entry in ff.get_audit_log(5): print(f" [{entry.timestamp[:19]}] {entry.action}: {entry.flag_name}") elif args.command == "list": ff = get_flags() flags = ff.list_flags() print(f"Feature Flags ({len(flags)}):") for flag in flags: status = "✓" if flag.enabled else "✗" print(f" [{status}] {flag.name}: {flag.description or 'No description'}") elif args.command == "status": ff = get_flags() print(json.dumps(ff.get_status(), indent=2)) if __name__ == "__main__": main()