Deploying code and releasing features don’t have to be the same thing. Feature flags let you ship code to production while controlling who sees it, when they see it, and how quickly you roll it out. This separation is transformative for both velocity and safety.

Why Feature Flags?

Traditional deployment: code goes live, everyone gets it immediately, and if something breaks, you redeploy.

With feature flags:

  • Deploy anytime — code exists in production but isn’t active
  • Release gradually — 1% of users, then 10%, then 50%, then everyone
  • Instant rollback — flip a switch, no deployment needed
  • Test in production — real traffic, real conditions, controlled exposure
  • A/B testing — compare variants with actual user behavior

Basic Implementation

Start simple. A feature flag is just a conditional:

1
2
3
4
5
6
7
8
9

# config.py
FEATURE_FLAGS = {
    'new_checkout_flow': False,
    'dark_mode': True,
    'ai_recommendations': False,
}

def is_enabled(flag_name: str, default: bool = False) -> bool:
    return FEATURE_FLAGS.get(flag_name, default)

1
2
3
4
5
6
7

# views.py
from config import is_enabled

def checkout(request):
    if is_enabled('new_checkout_flow'):
        return new_checkout_handler(request)
    return legacy_checkout_handler(request)

This works, but requires redeployment to change flags. Let’s make it dynamic.

Dynamic Flags with Redis

Store flags externally so you can change them without deploying:

 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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

import redis
import json
from functools import lru_cache
from typing import Any, Optional

class FeatureFlags:
    def __init__(self, redis_client: redis.Redis, prefix: str = "ff"):
        self.redis = redis_client
        self.prefix = prefix
        self._local_cache = {}
        self._cache_ttl = 30  # seconds
    
    def _key(self, flag: str) -> str:
        return f"{self.prefix}:{flag}"
    
    def set_flag(self, flag: str, config: dict):
        """
        Config example:
        {
            "enabled": True,
            "percentage": 100,
            "allowed_users": ["user123"],
            "allowed_groups": ["beta_testers"]
        }
        """
        self.redis.set(self._key(flag), json.dumps(config))
    
    def get_flag_config(self, flag: str) -> Optional[dict]:
        data = self.redis.get(self._key(flag))
        if data:
            return json.loads(data)
        return None
    
    def is_enabled(
        self, 
        flag: str, 
        user_id: Optional[str] = None,
        user_groups: Optional[list] = None,
        default: bool = False
    ) -> bool:
        config = self.get_flag_config(flag)
        
        if not config:
            return default
        
        if not config.get('enabled', False):
            return False
        
        # Check user allowlist
        if user_id and user_id in config.get('allowed_users', []):
            return True
        
        # Check group allowlist
        if user_groups:
            allowed_groups = set(config.get('allowed_groups', []))
            if allowed_groups & set(user_groups):
                return True
        
        # Percentage rollout
        percentage = config.get('percentage', 100)
        if percentage < 100:
            if not user_id:
                return False
            # Consistent hashing: same user always gets same result
            hash_val = hash(f"{flag}:{user_id}") % 100
            return hash_val < percentage
        
        return True


# Usage
redis_client = redis.Redis(host='localhost', port=6379)
flags = FeatureFlags(redis_client)

# Set up a gradual rollout
flags.set_flag('new_search', {
    'enabled': True,
    'percentage': 10,  # 10% of users
    'allowed_users': ['internal_tester_1'],
    'allowed_groups': ['employees']
})

# Check the flag
if flags.is_enabled('new_search', user_id=current_user.id, user_groups=current_user.groups):
    return new_search(query)
return legacy_search(query)

Progressive Rollout Strategy

Don’t flip flags from 0 to 100%. Use gradual rollouts:

 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

class RolloutManager:
    def __init__(self, flags: FeatureFlags):
        self.flags = flags
    
    def start_rollout(self, flag: str, stages: list[dict]):
        """
        stages = [
            {'percentage': 1, 'duration_hours': 2},
            {'percentage': 5, 'duration_hours': 4},
            {'percentage': 25, 'duration_hours': 12},
            {'percentage': 50, 'duration_hours': 24},
            {'percentage': 100, 'duration_hours': None},  # Final
        ]
        """
        config = self.flags.get_flag_config(flag) or {}
        config['rollout_stages'] = stages
        config['rollout_started'] = time.time()
        config['current_stage'] = 0
        config['percentage'] = stages[0]['percentage']
        config['enabled'] = True
        self.flags.set_flag(flag, config)
    
    def advance_rollout(self, flag: str) -> bool:
        """Advance to next stage if duration elapsed. Returns True if advanced."""
        config = self.flags.get_flag_config(flag)
        if not config or 'rollout_stages' not in config:
            return False
        
        stages = config['rollout_stages']
        current = config.get('current_stage', 0)
        
        if current >= len(stages) - 1:
            return False  # Already at final stage
        
        # Check if ready to advance
        stage_start = config.get('stage_started', config['rollout_started'])
        duration = stages[current].get('duration_hours')
        
        if duration and (time.time() - stage_start) < duration * 3600:
            return False  # Not yet
        
        # Advance
        next_stage = current + 1
        config['current_stage'] = next_stage
        config['percentage'] = stages[next_stage]['percentage']
        config['stage_started'] = time.time()
        self.flags.set_flag(flag, config)
        
        return True
    
    def emergency_rollback(self, flag: str):
        """Instant disable."""
        config = self.flags.get_flag_config(flag) or {}
        config['enabled'] = False
        config['rolled_back_at'] = time.time()
        self.flags.set_flag(flag, config)

A/B Testing with Variants

Feature flags aren’t just on/off. Test multiple variants:

 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

class VariantFlag:
    def __init__(self, flags: FeatureFlags):
        self.flags = flags
    
    def set_experiment(self, flag: str, variants: dict[str, int]):
        """
        variants = {'control': 50, 'variant_a': 25, 'variant_b': 25}
        Percentages must sum to 100.
        """
        assert sum(variants.values()) == 100
        
        config = {
            'enabled': True,
            'type': 'experiment',
            'variants': variants
        }
        self.flags.set_flag(flag, config)
    
    def get_variant(self, flag: str, user_id: str) -> str:
        config = self.flags.get_flag_config(flag)
        
        if not config or not config.get('enabled'):
            return 'control'
        
        variants = config.get('variants', {'control': 100})
        
        # Consistent assignment
        hash_val = hash(f"{flag}:{user_id}") % 100
        
        cumulative = 0
        for variant, percentage in variants.items():
            cumulative += percentage
            if hash_val < cumulative:
                return variant
        
        return 'control'


# Usage
experiments = VariantFlag(flags)

experiments.set_experiment('pricing_page', {
    'control': 34,
    'higher_price': 33,
    'lower_price': 33
})

variant = experiments.get_variant('pricing_page', user_id)
if variant == 'higher_price':
    price = base_price * 1.2
elif variant == 'lower_price':
    price = base_price * 0.8
else:
    price = base_price

# Track which variant the user saw for analytics
analytics.track('pricing_shown', {'variant': variant, 'price': price})

FastAPI Integration

Middleware for automatic flag context:

 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

from fastapi import FastAPI, Request, Depends
from contextvars import ContextVar

app = FastAPI()
flags = FeatureFlags(redis.Redis())

# Context for current request's flag evaluation
flag_context: ContextVar[dict] = ContextVar('flag_context', default={})

@app.middleware("http")
async def flag_middleware(request: Request, call_next):
    # Extract user context for flag evaluation
    user_id = request.headers.get('X-User-ID')
    user_groups = request.headers.get('X-User-Groups', '').split(',')
    
    flag_context.set({
        'user_id': user_id,
        'user_groups': [g.strip() for g in user_groups if g.strip()]
    })
    
    response = await call_next(request)
    return response

def check_flag(flag_name: str, default: bool = False) -> bool:
    ctx = flag_context.get()
    return flags.is_enabled(
        flag_name,
        user_id=ctx.get('user_id'),
        user_groups=ctx.get('user_groups'),
        default=default
    )

# Usage in routes
@app.get("/search")
async def search(q: str):
    if check_flag('new_search_algorithm'):
        return await new_search(q)
    return await legacy_search(q)

Cleanup: Don’t Let Flags Accumulate

Technical debt warning: old flags pile up. Track and remove them:

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

class FlagLifecycle:
    def __init__(self, flags: FeatureFlags):
        self.flags = flags
    
    def get_stale_flags(self, days: int = 30) -> list[str]:
        """Find flags that have been at 100% for over N days."""
        stale = []
        
        for key in self.flags.redis.scan_iter(f"{self.flags.prefix}:*"):
            flag = key.decode().split(':')[1]
            config = self.flags.get_flag_config(flag)
            
            if not config:
                continue
            
            # Flag is fully rolled out
            if config.get('percentage', 0) == 100 and config.get('enabled'):
                rolled_out_at = config.get('stage_started', config.get('rollout_started', 0))
                if rolled_out_at and (time.time() - rolled_out_at) > days * 86400:
                    stale.append(flag)
        
        return stale

When a flag has been at 100% for a month, it’s time to remove the conditional and delete the flag. The feature is now just… the code.

Best Practices

  1. Name flags descriptively: enable_new_checkout_v2 not flag_123
  2. Default to off: New flags should be disabled until explicitly enabled
  3. Use consistent hashing: Same user should always see the same variant
  4. Log flag evaluations: For debugging and analytics
  5. Set expiration expectations: Every flag should have an expected removal date
  6. Keep flag logic simple: Complex conditions belong in code, not flag config

Feature flags transform how you ship software. Start with simple boolean flags, add percentage rollouts when you need them, and build toward full experimentation infrastructure as your needs grow.