How to Build a Feature Flag System in Python With Sticky Percentage Rollouts
A step-by-step Python tutorial that builds a feature flag system from scratch, covering sticky percentage rollouts, targeting overrides, a kill switch, and the real bugs that break naive...
Imagine your team is about to ship a rewritten pricing engine. The new code passed every test in staging, but staging never sees real traffic patterns, real payment data, or real edge cases. You do not want to flip every customer over to it at once and find out about a bug from your support queue. You want to turn it on for 10% of customers, watch what happens, then dial it up to 50%, then 100%, and if anything looks wrong, turn it off in seconds, not by rolling back a deployment. That is what a feature flag system gives you, and by the end of this tutorial you will have built a real one from scratch in Python.
Table Of Content
- What You Will Build
- Prerequisites
- Step 1: Why a Hardcoded Constant Is Not Enough
- Step 2: A Configuration-Driven Flag Store
- Step 3: Percentage Rollouts, and a Bug Hiding in the Obvious Approach
- Step 4: Fix It With Deterministic, Sticky Bucketing
- Why hash() Does Not Work Here
- Step 5: The Property That Makes Gradual Rollouts Safe
- Step 6: Targeting Overrides
- Common Mistake: Checking Overrides in the Wrong Order
- Step 7: The Kill Switch
- Step 8: The Config You Never Reload Might as Well Not Exist
- Step 9: Putting It All Together
- Testing Feature-Flagged Code
- Common Mistakes and Gotchas, Recap
- How to Verify Everything Works End to End
- Next Steps
A feature flag (also called a feature toggle) is a piece of configuration, checked at runtime, that decides which code path runs for a given request or user. The code for both the old and new behavior already exists in your deployed application; the flag just decides which one actually executes right now. Because the decision lives in configuration instead of in which version of the code is deployed, you can change it instantly, without a new deployment, and you can change it for a subset of users instead of everyone at once. That single idea, separating “the code is deployed” from “the feature is turned on,” is what makes gradual rollouts, kill switches, and A/B tests possible.
This is not a niche technique. OpenFeature, a vendor-neutral specification for feature flag evaluation that is, as of this writing, a Cloud Native Computing Foundation incubating project, describes feature flags as letting teams “reduce the need for long-running feature branches,” “perform canary releases,” “perform A/B testing,” and “safely degrade parts of a production system that are experiencing an outage.” Those four use cases map almost exactly to the four pieces this tutorial builds: a dynamic on/off switch, a percentage rollout, targeting overrides, and a kill switch.
What You Will Build
You will build a small, self-contained feature flag library with no external services and no third-party packages: just the Python standard library plus pytest for the test suite. Along the way you will personally reproduce three real bugs that trip up naive implementations of exactly this kind of system, see the actual broken output each one produces, and then fix each one:
- A percentage rollout built on
random.random()that flips the same user on and off on every single request. - A rollout that reshuffles which users are enabled every time you raise the percentage, instead of only ever adding more users.
- A configuration store that silently ignores every change you make to its config file after the process starts.
By the last step you will have a Flag class with a kill switch, per-user overrides, and sticky percentage rollouts; a FlagStore that reloads its configuration on a schedule; and a test suite that verifies all of it without needing a database, a network call, or a running server.
Prerequisites
- Python 3.10 or newer (this tutorial was written and personally tested against Python 3.13.14, but nothing here depends on a feature newer than 3.10).
- Comfort with functions, classes, and reading a dict. No prior experience with feature flags, hashing, or distributed systems is assumed; every term is defined the first time it is used.
pytestif you want to run the verification suite at the end (pip install pytest). Everything else uses only the standard library:hashlib,json,pathlib, andtime.- No accounts, servers, or paid services of any kind. Every example below runs as a plain local script.
Step 1: Why a Hardcoded Constant Is Not Enough
The most naive version of a feature flag is a constant sitting near the top of a file:
USE_NEW_PRICING_ENGINE = False
def calculate_price(base_price):
if USE_NEW_PRICING_ENGINE:
return new_pricing_engine(base_price)
return old_pricing_engine(base_price)
This already buys you something: the old and new code paths both exist in the deployed application, chosen by one variable. But it has three real limits. First, flipping it requires editing the file and shipping a new deployment, so “turn it off right now” is only as fast as your deploy pipeline. Second, it is all-or-nothing: every request gets the same answer, so there is no way to try it on a slice of traffic first. Third, testing the “off” branch in production means either commenting the line back and forth or maintaining two deployed branches, exactly the branching pain feature flags exist to avoid. The rest of this tutorial fixes these one at a time.
Step 2: A Configuration-Driven Flag Store
The first fix is to stop hardcoding the decision in code and start reading it from a small piece of configuration that lives outside your deployed application, in this case a JSON file. This is the same idea behind what Pete Hodgson, writing on Martin Fowler’s site in his widely cited article on Feature Toggles, calls a Toggle Router: a small component whose only job is to answer “is this flag on,” decoupled from wherever that answer actually comes from.
import json
import pathlib
class FlagStore:
def __init__(self, config_path):
self.config_path = pathlib.Path(config_path)
self._flags = {}
self.reload()
def reload(self):
with open(self.config_path, "r", encoding="utf-8") as f:
self._flags = json.load(f)
def is_enabled(self, flag_name):
flag = self._flags.get(flag_name)
if flag is None:
return False
return bool(flag.get("enabled", False))
if __name__ == "__main__":
config = {"new_pricing_engine": {"enabled": False}}
with open("flags.json", "w", encoding="utf-8") as f:
json.dump(config, f)
store = FlagStore("flags.json")
print("Before flip:", store.is_enabled("new_pricing_engine"))
# Flip the flag by editing the config file on disk, no code change, no redeploy
config["new_pricing_engine"]["enabled"] = True
with open("flags.json", "w", encoding="utf-8") as f:
json.dump(config, f)
store.reload()
print("After flip:", store.is_enabled("new_pricing_engine"))
Running it produces:
Before flip: False
After flip: True
Nothing about the Python code changed between those two lines of output; only the contents of flags.json changed, and the store picked it up when reload() ran. That is the whole point: the decision now lives in data, not in a deployed constant, which is what makes everything in the next several steps possible.
Step 3: Percentage Rollouts, and a Bug Hiding in the Obvious Approach
A binary on/off switch is a start, but the real goal from the introduction was “10% of customers, then 50%, then 100%.” The most obvious way to write that is to roll a die on every request:
import random
class NaivePercentageFlag:
"""Rolls a feature out to a percentage of traffic using plain randomness."""
def __init__(self, percentage):
self.percentage = percentage
def is_enabled(self, user_id):
# user_id is accepted but never used -- that's the bug
return random.random() * 100 < self.percentage
if __name__ == "__main__":
flag = NaivePercentageFlag(percentage=25)
print("Same user, 20 consecutive checks, 25% rollout:")
results = [flag.is_enabled("alice") for _ in range(20)]
print(results)
print(f"alice was enabled on {sum(results)} of 20 checks")
Run that and watch what happens to a single user, Alice, across 20 consecutive checks:
Same user, 20 consecutive checks, 25% rollout:
[True, False, False, False, True, False, False, False, False, False, False, True, False, False, False, True, False, False, False, False]
alice was enabled on 4 of 20 checks
Alice is enabled, then disabled, then enabled again, at random, on every single check. In a real application this is a genuinely bad experience: a UI element that appears and disappears on every page reload, an API response whose shape changes from one request to the next for the same customer, or a pricing calculation that gives the same order two different totals depending on exactly when it is computed. The rollout percentage is correct in aggregate (roughly a quarter of checks came back True), but it is completely wrong per user, because is_enabled never looks at user_id at all. Two checks for the same person are two independent coin flips.
Step 4: Fix It With Deterministic, Sticky Bucketing
What you actually want is for the same user to get the same answer every time, while the aggregate percentage across all users still matches the configured rollout. The standard fix is to replace randomness with a deterministic hash of the flag name and the user id: compute a number between 0 and 99 from that combination, and compare it to the rollout percentage.
import hashlib
class StickyPercentageFlag:
"""Rolls a feature out to a percentage of traffic, deterministically per user."""
def __init__(self, flag_name, percentage):
self.flag_name = flag_name
self.percentage = percentage
def _bucket(self, user_id):
key = f"{self.flag_name}:{user_id}".encode("utf-8")
digest = hashlib.sha256(key).hexdigest()
# first 8 hex chars -> plenty of entropy, mod 100 gives a 0-99 bucket
return int(digest[:8], 16) % 100
def is_enabled(self, user_id):
return self._bucket(user_id) < self.percentage
if __name__ == "__main__":
flag = StickyPercentageFlag("new_pricing_engine", percentage=25)
print("Same user, 20 consecutive checks, 25% rollout:")
results = [flag.is_enabled("alice") for _ in range(20)]
print(results)
print(f"alice's bucket: {flag._bucket('alice')}")
Output:
Same user, 20 consecutive checks, 25% rollout:
[False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False, False]
alice's bucket: 27
Alice now gets the exact same answer on all 20 checks, because her bucket (27) never changes and is always compared against the same threshold (25). No database row or session cookie is needed to remember her assignment; the hash itself is the memory. Run this with 20 different users at the same 25% setting and roughly a quarter of them land under the threshold, but each individual user's result stays fixed no matter how many times you ask.
Why hash() Does Not Work Here
It is tempting to reach for Python's built-in hash() function instead of hashlib.sha256, since it is shorter to write. Do not: as of Python 3.3, hash randomization is enabled by default, so the __hash__() values of str and bytes objects are "salted with an unpredictable random value" that is "constant within an individual Python process" but "not predictable between repeated invocations of Python." Running python -c "print(hash('alice'))" three separate times on this machine produced three completely different numbers:
-4902057457378280476
2949837854614999797
-7902934958173389133
If you bucketed users with hash(), every single deploy or restart of your application would silently reshuffle every user into a different bucket, since each new process gets a fresh random seed. hashlib.sha256, by contrast, always produces the same digest for the same input, in any process, on any machine, forever. That stability is the entire mechanism this system depends on.
Step 5: The Property That Makes Gradual Rollouts Safe
There is a second, less obvious requirement for a rollout to be trustworthy: raising the percentage from 10% to 20% should only ever add new users to the enabled group, never remove anyone who was already enabled. If it removed people, "gradually roll out" would really mean "randomly reshuffle who has the feature every time you touch the dial," which defeats the purpose of a gradual rollout entirely. The hash-bucket approach from Step 4 gives you this for free, because it compares the same fixed bucket against a growing threshold. This is worth proving with real numbers instead of trusting it by assumption.
from step4_hash_bucketing import StickyPercentageFlag
FLAG_NAME = "new_pricing_engine"
USER_COUNT = 5000
PERCENTAGES = [10, 20, 50, 80, 100]
def enabled_set(percentage):
flag = StickyPercentageFlag(FLAG_NAME, percentage)
return {f"user_{i}" for i in range(USER_COUNT) if flag.is_enabled(f"user_{i}")}
if __name__ == "__main__":
sets_by_pct = {pct: enabled_set(pct) for pct in PERCENTAGES}
for pct in PERCENTAGES:
print(f"{pct:3d}%: {len(sets_by_pct[pct])} of {USER_COUNT} users enabled "
f"({len(sets_by_pct[pct]) / USER_COUNT * 100:.1f}%)")
print()
print("Checking the monotonic growth property (every user enabled at pct P")
print("is still enabled at every pct > P):")
all_hold = True
for i in range(len(PERCENTAGES) - 1):
lower_pct, higher_pct = PERCENTAGES[i], PERCENTAGES[i + 1]
lower_set, higher_set = sets_by_pct[lower_pct], sets_by_pct[higher_pct]
is_subset = lower_set.issubset(higher_set)
removed = lower_set - higher_set
print(f" {lower_pct}% subset of {higher_pct}%? {is_subset} "
f"(users removed going from {lower_pct}% to {higher_pct}%: {len(removed)})")
all_hold = all_hold and is_subset
print()
print(f"Monotonic growth holds across all tested tiers: {all_hold}")
With 5,000 simulated users:
10%: 507 of 5000 users enabled (10.1%)
20%: 1047 of 5000 users enabled (20.9%)
50%: 2523 of 5000 users enabled (50.5%)
80%: 3990 of 5000 users enabled (79.8%)
100%: 5000 of 5000 users enabled (100.0%)
Checking the monotonic growth property (every user enabled at pct P
is still enabled at every pct > P):
10% subset of 20%? True (users removed going from 10% to 20%: 0)
20% subset of 50%? True (users removed going from 20% to 50%: 0)
50% subset of 80%? True (users removed going from 50% to 80%: 0)
80% subset of 100%? True (users removed going from 80% to 100%: 0)
Monotonic growth holds across all tested tiers: True
Every one of the 507 users enabled at 10% is still enabled at 20%, at 50%, at 80%, and at 100%. Zero users were removed at any step. The observed percentages (10.1%, 20.9%, 50.5%, 79.8%) also track the configured targets closely, which is exactly the law of large numbers at work: with only a handful of users the match would be noisier, but with 5,000 it converges tightly.
Common mistake: this guarantee only holds if the input to the hash stays fixed across rollout stages. A tempting but broken variation is to fold the percentage itself into the hash key, on the theory that it adds "extra randomness":
def broken_enabled_set(percentage):
import hashlib
enabled = set()
for i in range(USER_COUNT):
user_id = f"user_{i}"
# BUG: including `percentage` in the hash key means every rollout
# stage gets an entirely different bucket assignment
key = f"{FLAG_NAME}:{percentage}:{user_id}".encode("utf-8")
bucket = int(hashlib.sha256(key).hexdigest()[:8], 16) % 100
if bucket < percentage:
enabled.add(user_id)
return enabled
print("Contrast: a broken version that hashes in the percentage itself")
broken_10 = broken_enabled_set(10)
broken_20 = broken_enabled_set(20)
still_in = broken_10.issubset(broken_20)
churned = broken_10 - broken_20
print(f" broken 10% subset of broken 20%? {still_in} "
f"(users who churned OUT despite the rollout growing: {len(churned)})")
Running the broken version:
Contrast: a broken version that hashes in the percentage itself
broken 10% subset of broken 20%? False (users who churned OUT despite the rollout growing: 404)
Of the 507 users enabled at the broken version's 10% tier, 404 of them (about 80%) got a completely different bucket once the percentage changed to 20 and lost access, even though the rollout was supposedly growing. The lesson is not "hashing is fragile"; it is that the hash's input must be exactly the stable identity you want to bucket by (here, the flag name and the user id) and nothing that changes between rollout stages.
Step 6: Targeting Overrides
Percentage rollouts alone are not enough for two situations that come up constantly in practice: your own QA team wants the new feature on right now regardless of their random bucket, and one specific customer is having a bad time with it and needs it off immediately while everyone else stays on. Both are handled by checking an explicit override list before falling back to the percentage rule.
import hashlib
class Flag:
"""
A single feature flag with:
- a global kill switch
- explicit per-user overrides (checked before the percentage rule)
- a percentage rollout using sticky hash bucketing
"""
def __init__(self, name, config):
self.name = name
self.config = config
def _bucket(self, user_id):
key = f"{self.name}:{user_id}".encode("utf-8")
digest = hashlib.sha256(key).hexdigest()
return int(digest[:8], 16) % 100
def is_enabled(self, user_id):
if not self.config.get("enabled", True):
return False # kill switch: overrides everything else
overrides = self.config.get("overrides", {})
if user_id in overrides.get("disabled_for", []):
return False
if user_id in overrides.get("enabled_for", []):
return True
percentage = self.config.get("percentage", 0)
return self._bucket(user_id) < percentage
if __name__ == "__main__":
config = {
"enabled": True,
"percentage": 0, # nobody gets it via the rollout yet
"overrides": {
"enabled_for": ["internal_qa_1", "internal_qa_2"],
"disabled_for": ["problem_user_42"],
},
}
flag = Flag("new_pricing_engine", config)
print("percentage=0, QA users force-enabled, one user force-disabled:")
for user_id in ["internal_qa_1", "internal_qa_2", "problem_user_42", "random_user_9"]:
print(f" {user_id}: {flag.is_enabled(user_id)}")
print()
print("Now bump the rollout to 100% -- overrides still apply first:")
config["percentage"] = 100
for user_id in ["problem_user_42", "random_user_9"]:
print(f" {user_id}: {flag.is_enabled(user_id)}")
Output:
percentage=0, QA users force-enabled, one user force-disabled:
internal_qa_1: True
internal_qa_2: True
problem_user_42: False
random_user_9: False
Now bump the rollout to 100% -- overrides still apply first:
problem_user_42: False
random_user_9: True
Even at a 0% rollout, both QA accounts see the feature, because enabled_for is checked before the percentage math ever runs. And even once the rollout reaches 100%, problem_user_42 stays off, because disabled_for is also checked before the percentage math. The order these checks happen in is not a style choice, it is a correctness requirement, and the next section shows exactly what breaks if you get it wrong.
Common Mistake: Checking Overrides in the Wrong Order
Here is the same idea with the checks reordered: the percentage rule runs first and returns immediately on a match, before the code ever looks at the override list.
import hashlib
class WrongOrderFlag:
"""BUG: checks the percentage rollout BEFORE the disabled_for override."""
def __init__(self, name, config):
self.name = name
self.config = config
def _bucket(self, user_id):
key = f"{self.name}:{user_id}".encode("utf-8")
return int(hashlib.sha256(key).hexdigest()[:8], 16) % 100
def is_enabled(self, user_id):
percentage = self.config.get("percentage", 0)
if self._bucket(user_id) < percentage:
return True # returns early -- never reaches the override check below
overrides = self.config.get("overrides", {})
if user_id in overrides.get("disabled_for", []):
return False
return False
if __name__ == "__main__":
config = {
"percentage": 100,
"overrides": {"disabled_for": ["problem_user_42"]},
}
flag = WrongOrderFlag("new_pricing_engine", config)
print("percentage=100, problem_user_42 is explicitly on the disabled_for list:")
print(f" problem_user_42 enabled: {flag.is_enabled('problem_user_42')}"
" <-- override was silently ignored")
percentage=100, problem_user_42 is explicitly on the disabled_for list:
problem_user_42 enabled: True <-- override was silently ignored
At a 100% rollout, every user's bucket is less than 100, so the percentage check returns True immediately and the function exits before it ever reaches the disabled_for check a few lines later. The override is not wrong, it is simply unreachable code once the rollout is high enough. This is exactly the kind of bug that hides during a small, low-percentage rollout and only shows up once you are confident enough to push toward 100%, which is the worst possible time to discover it.
Step 7: The Kill Switch
Fowler and Hodgson's article categorizes this specific use as an Ops Toggle: a flag whose entire job is to let an operator degrade or disable a piece of functionality during an incident, without waiting on a deployment. It needs to override absolutely everything else, including the QA overrides from the previous step, because during an incident you want the feature off for everyone, no exceptions.
from step6_overrides import Flag
if __name__ == "__main__":
config = {
"enabled": True,
"percentage": 100,
"overrides": {"enabled_for": ["internal_qa_1"], "disabled_for": []},
}
flag = Flag("new_pricing_engine", config)
users = ["internal_qa_1", "random_user_1", "random_user_2"]
print("100% rollout, everyone enabled:")
for u in users:
print(f" {u}: {flag.is_enabled(u)}")
print()
print("Incident! Flip the global kill switch (no redeploy, just a config edit):")
config["enabled"] = False
for u in users:
print(f" {u}: {flag.is_enabled(u)}")
Output:
100% rollout, everyone enabled:
internal_qa_1: True
random_user_1: True
random_user_2: True
Incident! Flip the global kill switch (no redeploy, just a config edit):
internal_qa_1: False
random_user_1: False
random_user_2: False
Flipping config["enabled"] to False disables the feature for every single user, including the QA account that was explicitly force-enabled a moment earlier. This is why the kill switch check is the very first line in is_enabled: during an incident, "off for everyone right now" needs to win against every other rule in the system, not just the default one.
Step 8: The Config You Never Reload Might as Well Not Exist
Every step so far assumed the flag store always has current data. In a real, long-running application, that is not automatic. If your FlagStore reads its config file once, at startup, and never again, then editing that file while the process is running has no effect until the process restarts, quietly defeating the entire "no redeploy needed" premise this tutorial started with.
import json
import time
import pathlib
def write_config(percentage):
with open("flags_step8.json", "w", encoding="utf-8") as f:
json.dump({"new_pricing_engine": {"enabled": True, "percentage": percentage}}, f)
class LoadOnceFlagStore:
"""BUG: reads the config file exactly once, at construction time."""
def __init__(self, config_path):
with open(config_path, "r", encoding="utf-8") as f:
self._flags = json.load(f)
def get_percentage(self, flag_name):
return self._flags.get(flag_name, {}).get("percentage", 0)
class RefreshingFlagStore:
"""FIX: re-reads the config file if it's older than ttl_seconds."""
def __init__(self, config_path, ttl_seconds=1.0):
self.config_path = pathlib.Path(config_path)
self.ttl_seconds = ttl_seconds
self._flags = {}
self._loaded_at = 0.0
self._load()
def _load(self):
with open(self.config_path, "r", encoding="utf-8") as f:
self._flags = json.load(f)
self._loaded_at = time.monotonic()
def _maybe_refresh(self):
if time.monotonic() - self._loaded_at >= self.ttl_seconds:
self._load()
def get_percentage(self, flag_name):
self._maybe_refresh()
return self._flags.get(flag_name, {}).get("percentage", 0)
write_config(percentage=10)
print("--- LoadOnceFlagStore (the bug) ---")
broken_store = LoadOnceFlagStore("flags_step8.json")
print(f"percentage at startup: {broken_store.get_percentage('new_pricing_engine')}")
write_config(percentage=50)
print("(ops team just edited flags_step8.json on disk to bump rollout to 50%)")
print(f"percentage after the edit: {broken_store.get_percentage('new_pricing_engine')}"
" <-- still reports the stale value, config change had zero effect")
print()
print("--- RefreshingFlagStore (the fix, ttl_seconds=1.0) ---")
write_config(percentage=10)
fresh_store = RefreshingFlagStore("flags_step8.json", ttl_seconds=1.0)
print(f"percentage at startup: {fresh_store.get_percentage('new_pricing_engine')}")
write_config(percentage=50)
print("(ops team edits the file again to bump rollout to 50%)")
print(f"percentage immediately after edit (ttl not expired yet): "
f"{fresh_store.get_percentage('new_pricing_engine')}")
time.sleep(1.1)
print(f"percentage after waiting past the {fresh_store.ttl_seconds}s ttl: "
f"{fresh_store.get_percentage('new_pricing_engine')}"
" <-- picked up the change, no restart needed")
Output:
--- LoadOnceFlagStore (the bug) ---
percentage at startup: 10
(ops team just edited flags_step8.json on disk to bump rollout to 50%)
percentage after the edit: 10 <-- still reports the stale value, config change had zero effect
--- RefreshingFlagStore (the fix, ttl_seconds=1.0) ---
percentage at startup: 10
(ops team edits the file again to bump rollout to 50%)
percentage immediately after edit (ttl not expired yet): 10
percentage after waiting past the 1.0s ttl: 50 <-- picked up the change, no restart needed
LoadOnceFlagStore never notices the second write to the file at all; it keeps returning 10 forever, because it only ever opened the file the moment it was constructed. RefreshingFlagStore fixes this by tracking when it last loaded and re-reading the file once ttl_seconds has elapsed, using time.monotonic() rather than a wall-clock time so the comparison cannot be thrown off by a system clock adjustment. Notice that the fix still briefly returns the old value immediately after the edit, before the TTL expires; that staleness window is a real, deliberate tradeoff between "always current" (which would mean hitting the disk on every single check) and "eventually current." For a kill switch specifically, you generally want that window to be short, seconds rather than minutes, since the whole point of a kill switch is speed.
Step 9: Putting It All Together
The final module combines everything: the Flag class from Steps 6 and 7 with the kill switch checked first, then overrides, then the sticky percentage rule from Step 4, wired into a FlagStore that refreshes from disk on a TTL like Step 8.
# flags.py
import hashlib
import json
import pathlib
import time
class Flag:
def __init__(self, name, config):
self.name = name
self.config = config
def _bucket(self, user_id):
key = f"{self.name}:{user_id}".encode("utf-8")
digest = hashlib.sha256(key).hexdigest()
return int(digest[:8], 16) % 100
def is_enabled(self, user_id):
if not self.config.get("enabled", True):
return False
overrides = self.config.get("overrides", {})
if user_id in overrides.get("disabled_for", []):
return False
if user_id in overrides.get("enabled_for", []):
return True
percentage = self.config.get("percentage", 0)
return self._bucket(user_id) < percentage
class FlagStore:
def __init__(self, config_path, ttl_seconds=30.0):
self.config_path = pathlib.Path(config_path)
self.ttl_seconds = ttl_seconds
self._raw = {}
self._loaded_at = 0.0
self._load()
def _load(self):
with open(self.config_path, "r", encoding="utf-8") as f:
self._raw = json.load(f)
self._loaded_at = time.monotonic()
def _maybe_refresh(self):
if time.monotonic() - self._loaded_at >= self.ttl_seconds:
self._load()
def is_enabled(self, flag_name, user_id):
self._maybe_refresh()
flag_config = self._raw.get(flag_name, {"enabled": False})
return Flag(flag_name, flag_config).is_enabled(user_id)
And here is real application code using it, a pricing function that branches on the flag exactly the way Step 1 wanted to, except now it is dynamic, sticky per customer, override-aware, and instantly killable:
# pricing.py
def calculate_price(base_price, user_id, flags):
if flags.is_enabled("new_pricing_engine", user_id):
return round(base_price * 0.9, 2) # new engine: 10% off promo
return base_price # old engine: no discount
import json
from flags import FlagStore
from pricing import calculate_price
config = {
"new_pricing_engine": {
"enabled": True,
"percentage": 20,
"overrides": {"enabled_for": ["internal_qa_1"], "disabled_for": []},
}
}
with open("flags_e2e.json", "w", encoding="utf-8") as f:
json.dump(config, f)
store = FlagStore("flags_e2e.json", ttl_seconds=60.0)
customers = ["internal_qa_1"] + [f"customer_{i}" for i in range(30)]
discounted = 0
for c in customers:
price = calculate_price(50.0, c, store)
if price < 50.0:
discounted += 1
print(f" {c:15s} pays {price}")
print(f"\n{discounted} of {len(customers)} customers got the discounted price")
Real output from a 20% rollout against 31 customers (30 regular plus the forced QA override):
internal_qa_1 pays 45.0
customer_0 pays 50.0
customer_1 pays 50.0
customer_2 pays 50.0
customer_3 pays 50.0
customer_4 pays 50.0
customer_5 pays 50.0
customer_6 pays 50.0
customer_7 pays 50.0
customer_8 pays 50.0
customer_9 pays 50.0
customer_10 pays 50.0
customer_11 pays 50.0
customer_12 pays 50.0
customer_13 pays 50.0
customer_14 pays 50.0
customer_15 pays 50.0
customer_16 pays 50.0
customer_17 pays 50.0
customer_18 pays 50.0
customer_19 pays 50.0
customer_20 pays 50.0
customer_21 pays 45.0
customer_22 pays 50.0
customer_23 pays 50.0
customer_24 pays 50.0
customer_25 pays 50.0
customer_26 pays 45.0
customer_27 pays 50.0
customer_28 pays 45.0
customer_29 pays 50.0
4 of 31 customers got the discounted price
internal_qa_1 is discounted through the override, and 3 of the 30 regular customers (10%, a bit under the 20% target purely because 30 is a small sample, the same law-of-large-numbers effect from Step 5) landed in the enabled bucket on their own. Run it again with the same config and you will get the exact same 4 names every time, which is the entire point.
Testing Feature-Flagged Code
Hodgson's article also makes a point worth taking seriously: a feature-flagged codebase has more paths through it than an unflagged one, and both paths need real test coverage, not just the one that happens to be active today. The trick to testing code like calculate_price deterministically is dependency injection: pass in a fake flag provider that just returns whatever you tell it to, so your tests never depend on hashing, files, or randomness at all.
import json
import pytest
from flags import Flag, FlagStore
from pricing import calculate_price
class FakeFlags:
"""A test double: force any flag to True/False, no hashing or files involved."""
def __init__(self, enabled_flags=()):
self._enabled = set(enabled_flags)
def is_enabled(self, flag_name, user_id=None):
return flag_name in self._enabled
def test_calculate_price_when_flag_off():
flags = FakeFlags(enabled_flags=[])
assert calculate_price(100.0, "alice", flags) == 100.0
def test_calculate_price_when_flag_on():
flags = FakeFlags(enabled_flags=["new_pricing_engine"])
assert calculate_price(100.0, "alice", flags) == 90.0
def test_kill_switch_overrides_percentage():
flag = Flag("f", {"enabled": False, "percentage": 100})
assert flag.is_enabled("anyone") is False
def test_override_beats_percentage_rollout():
flag = Flag("f", {
"enabled": True,
"percentage": 0,
"overrides": {"enabled_for": ["vip_user"], "disabled_for": []},
})
assert flag.is_enabled("vip_user") is True
assert flag.is_enabled("random_user") is False
def test_disabled_for_beats_percentage_rollout():
flag = Flag("f", {
"enabled": True,
"percentage": 100,
"overrides": {"enabled_for": [], "disabled_for": ["blocked_user"]},
})
assert flag.is_enabled("blocked_user") is False
assert flag.is_enabled("anyone_else") is True
def test_percentage_rollout_is_sticky():
flag = Flag("f", {"enabled": True, "percentage": 37})
first = flag.is_enabled("stable_user")
for _ in range(50):
assert flag.is_enabled("stable_user") == first
def test_percentage_rollout_is_monotonic_across_two_stages():
users = [f"u{i}" for i in range(2000)]
low = Flag("f", {"enabled": True, "percentage": 15})
high = Flag("f", {"enabled": True, "percentage": 45})
low_enabled = {u for u in users if low.is_enabled(u)}
high_enabled = {u for u in users if high.is_enabled(u)}
assert low_enabled.issubset(high_enabled)
def test_flag_store_refreshes_after_ttl(tmp_path, monkeypatch):
import time as time_module
config_path = tmp_path / "flags.json"
config_path.write_text(json.dumps({"f": {"enabled": True, "percentage": 0}}))
fake_now = [0.0]
monkeypatch.setattr(time_module, "monotonic", lambda: fake_now[0])
store = FlagStore(config_path, ttl_seconds=10.0)
assert store.is_enabled("f", "anyone") is False
config_path.write_text(json.dumps({"f": {"enabled": True, "percentage": 100}}))
fake_now[0] = 5.0 # inside the ttl window
assert store.is_enabled("f", "anyone") is False
fake_now[0] = 11.0 # past the ttl window
assert store.is_enabled("f", "anyone") is True
Running pytest -v against it:
============================= test session starts =============================
platform win32 -- Python 3.13.14, pytest-9.1.1, pluggy-1.6.0
collected 8 items
test_flags.py::test_calculate_price_when_flag_off PASSED [ 12%]
test_flags.py::test_calculate_price_when_flag_on PASSED [ 25%]
test_flags.py::test_kill_switch_overrides_percentage PASSED [ 37%]
test_flags.py::test_override_beats_percentage_rollout PASSED [ 50%]
test_flags.py::test_disabled_for_beats_percentage_rollout PASSED [ 62%]
test_flags.py::test_percentage_rollout_is_sticky PASSED [ 75%]
test_flags.py::test_percentage_rollout_is_monotonic_across_two_stages PASSED [ 87%]
test_flags.py::test_flag_store_refreshes_after_ttl PASSED [100%]
============================== 8 passed in 0.07s ==============================
Notice the last test never actually sleeps for 10 real seconds; it uses monkeypatch to replace time.monotonic with a fake clock it controls directly, so the TTL logic from Step 8 can be verified instantly and deterministically instead of making the test suite slow and flaky.
Common Mistakes and Gotchas, Recap
- Using randomness instead of a stable hash.
random.random()makes the same user flip on and off on every check (Step 3). Bucket by a deterministic hash of the flag name and user id instead (Step 4). - Hashing with Python's built-in
hash(). String hashing is randomized per process by default, so bucket assignments would silently reshuffle on every restart. Usehashlib(Step 4). - Letting the rollout percentage (or anything else that changes over time) leak into the hash key. This breaks the monotonic growth guarantee and can churn the majority of your enabled users the moment you raise a rollout percentage (Step 5).
- Checking the percentage rule before the kill switch or overrides. A rule that returns early can make an emergency override or kill switch unreachable, and the bug only shows up once the rollout percentage is high enough to trigger it (Step 6, Step 7).
- Loading configuration once at startup and never refreshing it. Every edit to the flag configuration becomes a no-op until the next restart, which defeats the reason to use a flag system in the first place (Step 8).
- Testing only the "flag on" or only the "flag off" path. A flagged codebase has more branches than an unflagged one; use a fake flag provider to exercise both deterministically instead of relying on whatever the real rollout percentage happens to be during a test run (Testing section).
How to Verify Everything Works End to End
Before trusting a flag system like this one, confirm each of the following, ideally the same way this tutorial did: by running real code and reading the real output, not by inspecting the code and assuming it is correct.
- A flag flips when its configuration file changes and
reload()(or a TTL refresh) runs, with no code change (Step 2, Step 8). - The same user gets the same answer from a percentage rollout across many repeated checks (Step 4, and
test_percentage_rollout_is_sticky). - Raising a rollout percentage only adds users to the enabled set, never removes any (Step 5, and
test_percentage_rollout_is_monotonic_across_two_stages). - An explicit override wins over the percentage rule in both directions, force-enabled and force-disabled (Step 6, and both override tests).
- The kill switch disables the flag for every user, including anyone on an
enabled_foroverride list (Step 7, andtest_kill_switch_overrides_percentage). - A stale, un-refreshed config store does not pick up changes, and a TTL-based one does, after the TTL elapses (Step 8, and
test_flag_store_refreshes_after_ttl).
If every item above holds for your own copy of this code, and the full pytest run shows 8 passed, you have a feature flag system whose core guarantees you have personally verified rather than assumed.
Next Steps
The hash-based bucketing technique in this tutorial is the same underlying idea behind distributing keys evenly across nodes; if you want to go deeper on that mechanism specifically, How to Build Consistent Hashing in Python to Stop Cache Stampedes builds a full hash ring with virtual nodes from scratch. Feature flags are one member of a broader family of runtime-configurable resilience patterns covered elsewhere on this site: circuit breakers stop calling a dependency that is already down, bulkheads stop one slow dependency from starving every other request, and rate limiters control how much load a service accepts in the first place. All four exist to answer some version of the same question this tutorial started with: what should happen right now, without waiting for a new deployment.
From here, a natural next step is reading the rest of Pete Hodgson's Feature Toggles article, which covers categories of toggles this tutorial did not build (Release Toggles and Experiment Toggles for A/B testing), plus the "carrying cost" of toggles and why they should eventually be deleted once a rollout finishes. If you want to see how a production-grade, multi-language version of these same ideas is standardized, the OpenFeature specification defines the same evaluation-context and provider concepts this tutorial built informally, backed by SDKs for a long list of real languages and a choice of open-source or commercial backends.








No Comment! Be the first one.