Feature Flags: Deploy Without Fear

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

class FeatureFlags:
    def __init__(self):
        self.flags = {}
    
    def is_enabled(self, flag_name: str, user_id: str = None) -> bool:
        flag = self.flags.get(flag_name)
        if not flag:
            return False
        
        # Global kill switch
        if not flag.get("enabled", False):
            return False
        
        # Check user allowlist
        if user_id and user_id in flag.get("allowed_users", []):
            return True
        
        # Check percentage rollout
        if user_id and flag.get("percentage", 0) > 0:
            # Consistent hashing: same user always gets same result
            hash_val = hash(f"{flag_name}:{user_id}") % 100
            return hash_val < flag["percentage"]
        
        return flag.get("enabled", False)

flags = FeatureFlags()
flags.flags = {
    "new_checkout": {
        "enabled": True,
        "percentage": 10,  # 10% of users
        "allowed_users": ["user_123"]  # Plus specific users
    }
}

# Usage
if flags.is_enabled("new_checkout", user_id=current_user.id):
    return new_checkout_flow()
else:
    return old_checkout_flow()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# feature_flags.yaml
flags:
  new_checkout:
    enabled: true
    percentage: 10
    allowed_users:
      - user_123
      - user_456
    
  dark_mode:
    enabled: true
    percentage: 100  # Fully rolled out
    
  experimental_search:
    enabled: true
    percentage: 0
    allowed_users:
      - internal_tester_1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import yaml

class FeatureFlags:
    def __init__(self, config_path: str):
        with open(config_path) as f:
            self.config = yaml.safe_load(f)
        self.flags = self.config.get("flags", {})
    
    def reload(self):
        """Hot reload without restart"""
        with open(self.config_path) as f:
            self.config = yaml.safe_load(f)
        self.flags = self.config.get("flags", {})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

def rollout_schedule(flag_name: str):
    """
    Day 1: 1% (canary)
    Day 2: 10% 
    Day 3: 25%
    Day 4: 50%
    Day 5: 100%
    """
    pass

# Consistent user bucketing
import hashlib

def get_bucket(user_id: str, flag_name: str) -> int:
    """Returns 0-99, consistent for same user+flag"""
    key = f"{flag_name}:{user_id}"
    hash_bytes = hashlib.md5(key.encode()).digest()
    return int.from_bytes(hash_bytes[:2], 'big') % 100

def is_in_rollout(user_id: str, flag_name: str, percentage: int) -> bool:
    return get_bucket(user_id, flag_name) < percentage

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

from dataclasses import dataclass
from typing import List, Dict, Any

@dataclass
class TargetingRule:
    attribute: str  # e.g., "country", "plan", "user_id"
    operator: str   # e.g., "equals", "in", "greater_than"
    value: Any

class FeatureFlag:
    def __init__(self, name: str):
        self.name = name
        self.enabled = False
        self.rules: List[TargetingRule] = []
        self.percentage = 0
    
    def evaluate(self, context: Dict[str, Any]) -> bool:
        if not self.enabled:
            return False
        
        # Check targeting rules
        for rule in self.rules:
            if not self._evaluate_rule(rule, context):
                continue
            return True
        
        # Fall back to percentage
        user_id = context.get("user_id")
        if user_id and self.percentage > 0:
            return get_bucket(user_id, self.name) < self.percentage
        
        return False
    
    def _evaluate_rule(self, rule: TargetingRule, context: Dict) -> bool:
        value = context.get(rule.attribute)
        if value is None:
            return False
        
        if rule.operator == "equals":
            return value == rule.value
        elif rule.operator == "in":
            return value in rule.value
        elif rule.operator == "greater_than":
            return value > rule.value
        
        return False

# Usage
flag = FeatureFlag("premium_feature")
flag.enabled = True
flag.rules = [
    TargetingRule("plan", "in", ["pro", "enterprise"]),
    TargetingRule("country", "equals", "US"),
]
flag.percentage = 5  # 5% of remaining users

context = {
    "user_id": "user_123",
    "plan": "pro",
    "country": "US"
}

if flag.evaluate(context):
    show_premium_feature()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

# LaunchDarkly example
import ldclient
from ldclient.config import Config

ldclient.set_config(Config("sdk-key"))
client = ldclient.get()

user = {
    "key": "user_123",
    "email": "user@example.com",
    "custom": {
        "plan": "pro",
        "country": "US"
    }
}

if client.variation("new-checkout", user, False):
    return new_checkout()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

from datetime import datetime, timedelta

class FlagWithMetadata:
    def __init__(self, name: str):
        self.name = name
        self.created_at = datetime.now()
        self.last_checked = None
        self.check_count = 0
    
    def is_stale(self, days: int = 90) -> bool:
        return datetime.now() - self.created_at > timedelta(days=days)

# Alert on stale flags
for flag in all_flags:
    if flag.is_stale() and flag.percentage == 100:
        alert(f"Flag '{flag.name}' is fully rolled out and over 90 days old. Remove it.")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import pytest
from unittest.mock import patch

def test_new_checkout_enabled():
    with patch.object(flags, 'is_enabled', return_value=True):
        response = client.post("/checkout")
        assert response.json()["flow"] == "new"

def test_new_checkout_disabled():
    with patch.object(flags, 'is_enabled', return_value=False):
        response = client.post("/checkout")
        assert response.json()["flow"] == "old"

# Or use flag overrides in test config
@pytest.fixture
def feature_flags():
    return FeatureFlags(config={
        "new_checkout": {"enabled": True, "percentage": 100}
    })

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from prometheus_client import Counter

flag_evaluations = Counter(
    'feature_flag_evaluations_total',
    'Feature flag evaluations',
    ['flag_name', 'result']
)

def is_enabled(flag_name: str, user_id: str) -> bool:
    result = _evaluate_flag(flag_name, user_id)
    flag_evaluations.labels(flag_name, str(result)).inc()
    return result

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

KILL_SWITCHES = {
    "payments": True,      # Payments enabled
    "signups": True,       # Signups enabled
    "external_api": True,  # External API calls enabled
}

def process_payment(order):
    if not KILL_SWITCHES.get("payments", True):
        raise ServiceUnavailable("Payments temporarily disabled")
    
    return payment_processor.charge(order)

1
2
3
4

# ❌ Bad
if flags.is_enabled("feature_a"):
    if flags.is_enabled("feature_a_variant_2"):
        ...

1
2
3

# ❌ Flag from 2 years ago, nobody knows if it's safe to remove
if flags.is_enabled("new_user_flow_2022"):
    ...

1
2
3
4
5
6

# ✅ Good: Include ownership
new_checkout:
  enabled: true
  owner: "payments-team"
  expires: "2024-06-01"
  jira: "PAY-1234"