Introduction#
Feature flags (feature toggles) separate deployment from release. You deploy code to production with a flag disabled, then enable it when you are confident — for a subset of users, a percentage of traffic, or all users at once. This enables continuous deployment without continuous feature release, and dramatically simplifies rollback.
Flag Types#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Release flags: hide incomplete features in production
- Deployed but not exposed to users
- Removed after feature launches
Experiment flags: A/B testing
- User bucket determines which variant they see
- Measure metrics, pick winner, clean up
Ops flags: runtime configuration
- Kill switches for expensive features
- Emergency circuit breakers
- "Disable search indexing during migration"
Permission flags: gradual rollout
- Enable for beta users → 1% → 10% → 100%
- Canary release at the feature level
Simple In-Memory Implementation#
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
import hashlib
from typing import Any
class FeatureFlags:
def __init__(self, config: dict[str, Any]):
self._config = config
def is_enabled(self, flag: str, user_id: str | None = None) -> bool:
flag_config = self._config.get(flag)
if flag_config is None:
return False # unknown flag → disabled by default
if isinstance(flag_config, bool):
return flag_config
if isinstance(flag_config, dict):
return self._evaluate(flag_config, user_id)
return False
def _evaluate(self, config: dict, user_id: str | None) -> bool:
# Whitelist of specific users
if "users" in config and user_id in config["users"]:
return True
# Percentage rollout
if "percentage" in config and user_id is not None:
bucket = self._bucket(user_id)
return bucket < config["percentage"]
# Global on/off
return config.get("enabled", False)
def _bucket(self, user_id: str) -> float:
"""Deterministically assign user to 0-100 bucket."""
h = hashlib.md5(user_id.encode()).hexdigest()
return (int(h[:8], 16) / 0xFFFFFFFF) * 100
flags = FeatureFlags({
"new_search": {"percentage": 20}, # 20% of users
"dark_mode": {"users": ["user:1", "user:42"]}, # specific users
"maintenance": False, # globally off
"new_dashboard": True, # globally on
})
print(flags.is_enabled("new_search", "user:123")) # deterministic per user
LaunchDarkly / OpenFeature Integration#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import ldclient
from ldclient.config import Config as LDConfig
from openfeature import api
from openfeature.provider.launchdarkly_provider import LaunchDarklyProvider
# LaunchDarkly SDK
ldclient.set_config(LDConfig(sdk_key="sdk-your-key-here"))
client = ldclient.get()
def evaluate_flag(flag_key: str, user_id: str, default: bool = False) -> bool:
context = ldclient.Context.builder(user_id).kind("user").build()
return client.variation(flag_key, context, default)
def get_flag_variant(flag_key: str, user_id: str, default: str = "control") -> str:
context = ldclient.Context.builder(user_id).kind("user").build()
return client.variation(flag_key, context, default)
# Usage in a FastAPI handler
@app.get("/search")
async def search(query: str, current_user: User = Depends(get_current_user)):
if evaluate_flag("new-search-algorithm", current_user.id):
return await new_search(query)
return await legacy_search(query)
Database-Backed Feature Flags#
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
from sqlalchemy import Column, String, Boolean, Float, JSON
from sqlalchemy.ext.declarative import declarative_base
import json
Base = declarative_base()
class FeatureFlag(Base):
__tablename__ = "feature_flags"
name = Column(String, primary_key=True)
enabled = Column(Boolean, default=False)
rollout_percentage = Column(Float, nullable=True)
allowed_users = Column(JSON, default=list)
metadata = Column(JSON, default=dict)
class DatabaseFeatureFlags:
def __init__(self, db: Session, cache_ttl: float = 30.0):
self._db = db
self._cache: dict[str, tuple[Any, float]] = {}
self._cache_ttl = cache_ttl
def is_enabled(self, flag_name: str, user_id: str | None = None) -> bool:
flag = self._get_flag(flag_name)
if flag is None:
return False
if flag.allowed_users and user_id in flag.allowed_users:
return True
if flag.rollout_percentage is not None and user_id is not None:
bucket = self._bucket(user_id, flag_name)
return bucket < flag.rollout_percentage
return flag.enabled
def _get_flag(self, name: str) -> FeatureFlag | None:
import time
cached, ts = self._cache.get(name, (None, 0))
if time.time() - ts < self._cache_ttl:
return cached
flag = self._db.query(FeatureFlag).filter(FeatureFlag.name == name).first()
self._cache[name] = (flag, time.time())
return flag
def _bucket(self, user_id: str, flag_name: str) -> float:
h = hashlib.md5(f"{flag_name}:{user_id}".encode()).hexdigest()
return (int(h[:8], 16) / 0xFFFFFFFF) * 100
def set_rollout(self, flag_name: str, percentage: float) -> None:
flag = self._db.query(FeatureFlag).filter(FeatureFlag.name == flag_name).first()
if flag:
flag.rollout_percentage = percentage
self._db.commit()
self._cache.pop(flag_name, None) # invalidate cache
Flag Middleware for HTTP#
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
from fastapi import FastAPI, Request, Depends
from contextvars import ContextVar
_current_flags: ContextVar[dict] = ContextVar("current_flags", default={})
@app.middleware("http")
async def inject_feature_flags(request: Request, call_next):
user_id = get_user_id_from_request(request)
# Evaluate all flags for this user upfront
flags = {
"new_checkout": feature_flags.is_enabled("new_checkout", user_id),
"dark_mode": feature_flags.is_enabled("dark_mode", user_id),
"ai_recommendations": feature_flags.is_enabled("ai_recommendations", user_id),
}
_current_flags.set(flags)
response = await call_next(request)
# Add active flags to response headers for debugging
active = [k for k, v in flags.items() if v]
if active:
response.headers["X-Active-Flags"] = ",".join(active)
return response
def flag(name: str) -> bool:
return _current_flags.get().get(name, False)
@app.post("/checkout")
async def checkout(cart_id: str):
if flag("new_checkout"):
return await new_checkout_flow(cart_id)
return await legacy_checkout_flow(cart_id)
Gradual Rollout Pattern#
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
def graduated_rollout(
flag_name: str,
schedule: list[tuple[int, float]], # (day_offset, percentage)
start_date: str,
):
"""
Automatically increase rollout percentage over time.
Example schedule:
[(0, 1.0), (3, 5.0), (7, 25.0), (14, 50.0), (21, 100.0)]
Start at 1%, ramp to 100% over 3 weeks.
"""
from datetime import date, timedelta
start = date.fromisoformat(start_date)
today = date.today()
days_since_start = (today - start).days
target_percentage = 0.0
for day_offset, percentage in sorted(schedule):
if days_since_start >= day_offset:
target_percentage = percentage
current = get_flag_percentage(flag_name)
if target_percentage != current:
set_flag_percentage(flag_name, target_percentage)
logger.info(
"Auto-rollout %s: %.1f%% → %.1f%%",
flag_name, current, target_percentage
)
Cleanup: Removing Stale Flags#
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
# Stale feature flags are technical debt
# Track flag age and usage to identify cleanup candidates
import ast
import os
from pathlib import Path
def find_flag_usages(codebase_path: str, flag_name: str) -> list[str]:
"""Find all files that reference a feature flag."""
usages = []
for py_file in Path(codebase_path).rglob("*.py"):
content = py_file.read_text()
if flag_name in content:
usages.append(str(py_file))
return usages
# When removing a flag:
# 1. Determine winning variant (or that rollout is complete)
# 2. Replace flag check with the winning code path
# 3. Delete the flag from the database
# 4. Remove the flag evaluation code
# BAD: leaving dead code
if feature_flags.is_enabled("new_search"): # this flag is always True now
return new_search(query)
else:
return legacy_search(query) # dead code
# GOOD: clean up after rollout completes
return new_search(query) # flag removed, always takes the new path
Conclusion#
Feature flags enable continuous deployment by decoupling code deployment from feature release. Database-backed flags with caching provide the right balance of dynamic control and performance. Percentage-based rollouts limit blast radius for new features. The most important practice is flag cleanup — each flag left in place after its purpose is served adds cognitive overhead and potential bugs. Treat flag removal as part of the release process, not an afterthought.