| name | feature-flags |
| description | Feature flag implementation and management |
| license | MIT |
| compatibility | opencode |
| metadata | {"audience":"developers","category":"development"} |
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
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 []
.dependencies = .dependencies []
.metadata = .metadata {}
:
():
.storage = storage
.cache = {}
.cache_ttl =
() -> :
flag = ._get_flag(flag_key)
flag:
flag.enabled:
dep_key flag.dependencies:
.is_enabled(dep_key, user_id, context):
flag.target_users user_id flag.target_users:
flag.exclude_users user_id flag.exclude_users:
flag.percentage > :
._in_percentage(flag_key, user_id, flag.percentage)
strategy flag.strategies:
._evaluate_strategy(strategy, user_id, context):
flag.enabled
() -> [FeatureFlag]:
flag_key .cache:
cached_flag, timestamp = .cache[flag_key]
(datetime.utcnow() - timestamp).seconds < .cache_ttl:
cached_flag
flag = .storage.get_flag(flag_key)
flag:
.cache[flag_key] = (flag, datetime.utcnow())
flag
() -> :
user_id:
user_id =
hash_value = ()
bucket = hash_value %
bucket < percentage
() -> :
strategy_type = strategy.get()
strategy_type == :
user_id strategy.get(, [])
strategy_type == :
attr = strategy.get()
op = strategy.get()
value = strategy.get()
user_attr = context.get(attr) context
._compare(user_attr, op, value)
() -> :
operators = {
: a, e: a == e,
: a, e: a != e,
: a, e: e a a ,
: a, e: a.startswith(e) a ,
: a, e: a.endswith(e) a ,
: a, e: a > e a e ,
: a, e: a < e a e ,
}
op_func = operators.get(operator)
op_func:
op_func(actual, expected)
():
() -> :
():
engine = get_feature_flag_engine()
user_id = get_current_user_id()
engine.is_enabled(flag_key, user_id):
func(*args, **kwargs)
:
default
wrapper
decorator
():
process_new_checkout()
():
render_beta_dashboard()
:
():
.user_id = user_id
.attributes = attributes
() -> :
engine = get_feature_flag_engine()
engine.is_enabled(flag_key, .user_id, .attributes)
context = FeatureContext(
user_id=,
plan=,
region=,
age_days=
)
context.is_enabled():
show_new_feature()
:
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":"
assignment_key .user_assignments:
.user_assignments[assignment_key]
test = .tests[test_name]
variant = ._assign_variant(test_name, test, user_id)
.user_assignments[assignment_key] = variant
variant
() -> :
user_id:
user_id =
hash_value = ()
bucket = hash_value %
cumulative =
variant, weight (test.variants, test.weights):
cumulative += weight
bucket < cumulative:
variant
test.variants[-]
() -> :
variant = .get_variant(test_name, user_id)
variant:
analytics.track(
event=,
user_id=user_id,
properties={
: test_name,
: variant,
**(properties {})
}
)
() -> [, ]:
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