Python's expressiveness and rapid development cycle make it well-suited for building feature flag systems, particularly when flag evaluation needs to integrate with data pipelines, ML models, or analytics workflows. This guide covers implementing a complete feature flag architecture in Python, from evaluation engine to sync client and targeting rules.
Evaluation Engine
The core evaluator must be thread-safe (for WSGI servers like Gunicorn) and fast on the read path:
python
1import hashlib
2import struct
3import threading
4from dataclasses import dataclass, field
5from typing import Optional
6
7
8
9class Condition:
10 attribute: str
11 operator: str # eq, neq, in, contains, starts_with, gt, lt
12 values: list[str]
13
14
15
16class Rule:
17 conditions: list[Condition]
18 variant: str = ""
19 priority: int = 0
20
21
22
23class FlagConfig:
24 key: str
25 enabled: bool
26 percentage: float = 0.0
27 rules: list[Rule] = field(default_factory=list)
28 variants: dict[str, float] = field(default_factory=dict)
29 default_variant: str = ""
30
31
32
33class EvalContext:
34 user_id: str
35 email: str = ""
36 plan: str = ""
37 country: str = ""
38 properties: dict[str, str] = field(default_factory=dict)
39
40
41
42class EvalResult:
43 enabled: bool
44 variant: str = ""
45 reason: str = ""
46
47
48class FlagEvaluator:
49 def __init__(self):
50 self._flags: dict[str, FlagConfig] = {}
51 self._lock = threading.Lock()
52
53 def update(self, configs: list[FlagConfig]) -> None:
54 new_flags = {c.key: c for c in configs}
55 with self._lock:
56 self._flags = new_flags
57
58 def evaluate(self, flag_key: str, context: EvalContext) -> EvalResult:
59 flags = self._flags # Single read for consistency
60 flag = flags.get(flag_key)
61
62 if flag is None:
63 return EvalResult(enabled=False, reason="not_found")
64 if not flag.enabled:
65 return EvalResult(enabled=False, reason="disabled")
66
67 # Rule-based targeting
68 for rule in sorted(flag.rules, key=lambda r: r.priority, reverse=True):
69 if self._matches_all(rule.conditions, context):
70 return EvalResult(enabled=True, variant=rule.variant, reason="rule_match")
71
72 # Percentage rollout
73 if 0 < flag.percentage < 100:
74 bucket = self._hash_bucket(flag_key, context.user_id)
75 if bucket < flag.percentage:
76 return EvalResult(enabled=True, variant=flag.default_variant, reason="percentage")
77 return EvalResult(enabled=False, reason="percentage_excluded")
78
79 return EvalResult(enabled=True, variant=flag.default_variant, reason="default")
80
81 def is_enabled(self, flag_key: str, context: EvalContext) -> bool:
82 return self.evaluate(flag_key, context).enabled
83
84
85 def _hash_bucket(flag_key: str, user_id: str) -> float:
86 h = hashlib.sha256(f"{flag_key}:{user_id}".encode()).digest()
87 value = struct.unpack(">I", h[:4])[0]
88 return (value / 0xFFFFFFFF) * 100
89
90
91 def _matches_all(conditions: list[Condition], ctx: EvalContext) -> bool:
92 for cond in conditions:
93 value = _get_attribute(cond.attribute, ctx)
94 if not _match_condition(cond, value):
95 return False
96 return True
97
98
99def _get_attribute(attr: str, ctx: EvalContext) -> str:
100 mapping = {
101 "user_id": ctx.user_id,
102 "email": ctx.email,
103 "plan": ctx.plan,
104 "country": ctx.country,
105 }
106 return mapping.get(attr, ctx.properties.get(attr, ""))
107
108
109def _match_condition(cond: Condition, value: str) -> bool:
110 match cond.operator:
111 case "eq":
112 return len(cond.values) > 0 and value == cond.values[0]
113 case "neq":
114 return len(cond.values) > 0 and value != cond.values[0]
115 case "in":
116 return value in cond.values
117 case "contains":
118 return len(cond.values) > 0 and cond.values[0] in value
119 case "starts_with":
120 return len(cond.values) > 0 and value.startswith(cond.values[0])
121 case _:
122 return False
123
Async Sync Client
python
1import asyncio
2import httpx
3import logging
4
5logger = logging.getLogger(__name__)
6
7
8class FlagSyncClient:
9 def __init__(self, evaluator: FlagEvaluator, api_url: str, interval: float = 10.0):
10 self.evaluator = evaluator
11 self.api_url = api_url
12 self.interval = interval
13 self._etag: str | None = None
14 self._client = httpx.AsyncClient(timeout=5.0)
15
16 async def start(self):
17 await self._sync()
18 asyncio.create_task(self._sync_loop())
19
20 async def _sync_loop(self):
21 while True:
22 await asyncio.sleep(self.interval)
23 try:
24 await self._sync()
25 except Exception as e:
26 logger.error(f"Flag sync failed: {e}")
27
28 async def _sync(self):
29 headers = {}
30 if self._etag:
31 headers["If-None-Match"] = self._etag
32
33 response = await self._client.get(f"{self.api_url}/api/flags", headers=headers)
34
35 if response.status_code == 304:
36 return # No changes
37
38 response.raise_for_status()
39 configs = [FlagConfig(**f) for f in response.json()]
40 self.evaluator.update(configs)
41 self._etag = response.headers.get("etag")
42 logger.info(f"Synced {len(configs)} flags")
43
FastAPI Integration
python
1from fastapi import FastAPI, Request, Depends
2from contextlib import asynccontextmanager
3
4evaluator = FlagEvaluator()
5sync_client = FlagSyncClient(evaluator, "http://flag-service:8080")
6
7
8
9async def lifespan(app: FastAPI):
10 await sync_client.start()
11 yield
12
13
14app = FastAPI(lifespan=lifespan)
15
16
17def get_flag_context(request: Request) -> EvalContext:
18 return EvalContext(
19 user_id=request.headers.get("x-user-id", ""),
20 plan=request.headers.get("x-plan", ""),
21 country=request.headers.get("cf-ipcountry", ""),
22 )
23
24
25def require_flag(flag_key: str):
26 def dependency(ctx: EvalContext = Depends(get_flag_context)):
27 if not evaluator.is_enabled(flag_key, ctx):
28 raise HTTPException(status_code=404, detail="Feature not available")
29 return ctx
30 return Depends(dependency)
31
32
33
34async def checkout(request: CheckoutRequest):
35 return await process_checkout(request)
36
37
38
39async def dashboard(ctx: EvalContext = Depends(get_flag_context)):
40 features = {
41 "show_analytics": evaluator.is_enabled("analytics-dashboard", ctx),
42 "show_ai_insights": evaluator.is_enabled("ai-insights", ctx),
43 "show_export": evaluator.is_enabled("data-export", ctx),
44 }
45 return {"features": features, "data": await get_dashboard_data(features)}
46
Need a second opinion on your saas engineering architecture?
I run free 30-minute strategy calls for engineering teams tackling this exact problem.
Book a Free CallDjango Integration
python
1from django.utils.deprecation import MiddlewareMixin
2
3class FeatureFlagMiddleware(MiddlewareMixin):
4 def process_request(self, request):
5 user_id = getattr(request.user, "id", "") if request.user.is_authenticated else ""
6 request.flag_context = EvalContext(
7 user_id=str(user_id),
8 plan=getattr(request.user, "plan", ""),
9 )
10 request.flags = RequestFlags(evaluator, request.flag_context)
11
12
13class RequestFlags:
14 def __init__(self, evaluator: FlagEvaluator, context: EvalContext):
15 self._evaluator = evaluator
16 self._context = context
17 self._cache: dict[str, EvalResult] = {}
18
19 def is_enabled(self, flag_key: str) -> bool:
20 if flag_key not in self._cache:
21 self._cache[flag_key] = self._evaluator.evaluate(flag_key, self._context)
22 return self._cache[flag_key].enabled
23
24# In views
25def checkout_view(request):
26 if request.flags.is_enabled("new-checkout"):
27 return render(request, "checkout_v2.html")
28 return render(request, "checkout.html")
29
Decorator-Based Gating
python
1import functools
2
3def feature_gate(flag_key: str, fallback=None):
4 def decorator(func):
5
6 async def wrapper(*args, **kwargs):
7 request = kwargs.get("request") or (args[0] if args else None)
8 ctx = getattr(request, "flag_context", EvalContext(user_id=""))
9
10 if evaluator.is_enabled(flag_key, ctx):
11 return await func(*args, **kwargs)
12 if fallback:
13 return await fallback(*args, **kwargs)
14 raise HTTPException(status_code=404)
15 return wrapper
16 return decorator
17
18
19async def analytics_endpoint(request: Request):
20 return await generate_premium_analytics()
21
Testing
python
1import pytest
2
3class TestFlagEvaluator:
4 def test_percentage_rollout_distribution(self):
5 evaluator = FlagEvaluator()
6 evaluator.update([FlagConfig(key="test", enabled=True, percentage=50)])
7
8 enabled = sum(
9 1 for i in range(10_000)
10 if evaluator.is_enabled("test", EvalContext(user_id=f"user-{i}"))
11 )
12 assert 4800 <= enabled <= 5200
13
14 def test_sticky_bucketing(self):
15 evaluator = FlagEvaluator()
16 evaluator.update([FlagConfig(key="test", enabled=True, percentage=50)])
17
18 ctx = EvalContext(user_id="user-123")
19 first = evaluator.is_enabled("test", ctx)
20 assert all(evaluator.is_enabled("test", ctx) == first for _ in range(100))
21
22 def test_rule_targeting(self):
23 evaluator = FlagEvaluator()
24 evaluator.update([
25 FlagConfig(
26 key="enterprise-only",
27 enabled=True,
28 rules=[Rule(conditions=[Condition("plan", "in", ["enterprise"])])]
29 )
30 ])
31
32 assert evaluator.is_enabled("enterprise-only", EvalContext(user_id="1", plan="enterprise"))
33 assert not evaluator.is_enabled("enterprise-only", EvalContext(user_id="2", plan="free"))
34
Conclusion
Python's feature flag implementation trades raw evaluation speed for development velocity and integration flexibility. The evaluation engine handles thousands of evaluations per second — more than sufficient for web applications where each request evaluates 5-10 flags. The real advantage is how naturally Python flags integrate with data pipelines, ML inference, and analytics workflows where Python is already the dominant language.
The architecture follows the same principle as other languages: local evaluation with background sync. Python's threading.Lock provides safe concurrent access for WSGI servers, while the asyncio-based sync client integrates seamlessly with ASGI frameworks like FastAPI. The FastAPI dependency injection pattern provides particularly clean flag gating that keeps route handlers focused on business logic.