🚀 Rollout Control in Python: Build Your Own Feature Flag System

🧠 What Are Feature Flags?

Feature flags (also called feature toggles) allow you to turn features on or off at runtime, without redeploying or changing your codebase. They give you fine-grained control over how and when features are exposed to users, making them a powerful tool in modern software delivery.

🧩 How They Work (At a High Level)

You wrap feature logic in a conditional check like this:

if feature_flag_enabled("some_feature"):
    render_new_functionality()
else:
    render_old_functionality()

Instead of deploying new code to change behavior, you can just toggle a value in your database, config file, or dashboard.

🔄 Real Use Cases

🔧 What We’re Building

✅ Features:

• Create / Update / Read flags

• In-memory + JSON file persistence

• Boolean flags (on / off)

• Percentage-based rollout

• User-targeted flags

• REST API to manage flags

📐 System Architecture

Here’s the architecture we’ll use:

🔲 Architecture Components:

• FastAPI for REST endpoints

• Flag Store abstraction (JSON file)

• Rollout Engine (decides if a flag applies to a user)

🔧 Code Explanation: main.py

📄 File: main.py

This file contains the FastAPI web server that exposes the feature flag endpoints.

from fastapi import FastAPI, HTTPException
from flag_store import FlagStore
from rollout import RolloutEngine

• FastAPI: Web framework for building APIs in Python.

• HTTPException: Used to return errors like 404 if a flag isn’t found.

• FlagStore: Handles reading/writing flag data (from JSON or another backend).

• RolloutEngine: Evaluates if a flag is "on" for a specific user.

✅ App and Instances

app = FastAPI()
store = FlagStore("flags.json")
engine = RolloutEngine(store)

• FastAPI() initialize the app.

• FlagStore("flags.json") reads flag data from a file called flags.json.

• RolloutEngine contains logic to evaluate if a flag should be active for a user, based on flag rules like rollout %.

📥 GET: Check Flag Status

@app.get("/flags/{flag_name}")
def get_flag(flag_name: str, user_id: str = None):
    flag_result = engine.evaluate_flag(flag_name, user_id)
    if flag_result is None:
        raise HTTPException(status_code=404, detail="Flag not found")
    return {"flag": flag_name, "enabled": flag_result}

• GET /flags/<flag_name>: Fetches whether a feature is enabled for a user.

• Accepts optional query param user_id for personalized evaluation.

• Calls RolloutEngine.evaluate_flag() to calculate if the flag is active.

• Returns { "flag": "new_ui", "enabled": true }

📝 POST: Set a Flag

@app.post("/flags/{flag_name}")
def set_flag(flag_name: str, enabled: bool, rollout_percentage: int = 100):
    store.set_flag(flag_name, {"enabled": enabled, "rollout": rollout_percentage})
    return {"status": "updated"}

• POST /flags/<flag_name>: Updates (or creates) a flag.

• Query parameters:

  • enabled: whether the feature is on or off.

  • rollout_percentage: percentage of users to show the feature to (defaults to 100).

• Stores the data in the flag store as a dictionary:

  {
    "enabled": true,
    "rollout": 50
  }

📦 Code Explanation: flag_store.py

📄 File: flag_store.py

This file manages reading and writing flags. It simulates a key-value store like Redis or a database, using a local JSON file.

import json
from threading import Lock

• json: To serialize/deserialize flag data.

• Lock: Ensures thread-safe access to the flag store (for concurrent writes).

🏗️ Class: FlagStore

class FlagStore:
    def __init__(self, path):
        self.path = path
        self.lock = Lock()

• path: the path to a file (flags.json) where flags are saved.

• lock: ensures atomic writes (important if using this in multithreaded environments).

🔍 Load Flags on Startup

        try:
            with open(self.path, "r") as f:
                self.flags = json.load(f)
        except FileNotFoundError:
            self.flags = {}

• Tries to load existing flags from a file.

• If the file doesn’t exist, starts with an empty dictionary.

💾 Save to Disk

    def save(self):
        with open(self.path, "w") as f:
            json.dump(self.flags, f, indent=2)

• Write the current state of all flags to the file in a pretty format.

🧠 Access Methods

    def get_flag(self, name):
        return self.flags.get(name)

• Fetches a specific flag by name (e.g., "new_ui").

    def set_flag(self, name, value: dict):
        with self.lock:
            self.flags[name] = value
            self.save()

• Updates or creates a flag entry.

• Stores a dict-like { "enabled": true, "rollout": 50 }

• Uses lock to prevent race conditions while writing.

🧠 Code Explanation: rollout.py

This module contains the logic to decide whether a user sees a feature.

📄 File: rollout.py

import hashlib

• Uses hashing to assign users into “buckets” for percentage rollout.

🧠 Class: RolloutEngine

class RolloutEngine:
    def __init__(self, store):
        self.store = store

• Stores a reference to the FlagStore, so it can access flag configs.

🔍 Flag Evaluation Logic

    def evaluate_flag(self, name, user_id=None):
        flag = self.store.get_flag(name)
        if not flag:
            return None
        if not flag.get("enabled", False):
            return False

• Fetches the flag from the store.

• If the flag doesn’t exist → returns None (not found).

• If the flag is disabled, → immediately returns False.

🎯 Percentage Rollout Logic

        if not user_id:
            return flag["enabled"]

• If no user_id is given, just return the basic flag state.

        bucket = int(hashlib.sha256(user_id.encode()).hexdigest(), 16) % 100
        return bucket < flag.get("rollout", 100)

• Hashes the user_id to get a deterministic number from 0–99.

• Compares it to the rollout percentage.

• If the user falls within the bucket range (e.g., 0–49 for 50% rollout), the flag is active.

✅ This ensures the same user always gets the same flag state.

🧰 Complete code

A full example of a feature flag API using FastAPI, supporting boolean flags and gradual rollouts by user.

📁 Directory Structure

feature_flags/
├── main.py          # FastAPI app
├── flag_store.py    # Stores flags (JSON-based)
├── rollout.py       # Handles rollout logic
└── flags.json       # Stores flag data

1️⃣ main.py

from fastapi import FastAPI, HTTPException
from flag_store import FlagStore
from rollout import RolloutEngine

app = FastAPI()
store = FlagStore("flags.json")
engine = RolloutEngine(store)

@app.get("/flags/{flag_name}")
def get_flag(flag_name: str, user_id: str = None):
    flag_result = engine.evaluate_flag(flag_name, user_id)
    if flag_result is None:
        raise HTTPException(status_code=404, detail="Flag not found")
    return {"flag": flag_name, "enabled": flag_result}

@app.post("/flags/{flag_name}")
def set_flag(flag_name: str, enabled: bool, rollout_percentage: int = 100):
    store.set_flag(flag_name, {"enabled": enabled, "rollout": rollout_percentage})
    return {"status": "updated"}

2️⃣ flag_store.py

import json
from threading import Lock

class FlagStore:
    def __init__(self, path):
        self.path = path
        self.lock = Lock()
        try:
            with open(self.path, "r") as f:
                self.flags = json.load(f)
        except FileNotFoundError:
            self.flags = {}

    def save(self):
        with open(self.path, "w") as f:
            json.dump(self.flags, f, indent=2)

    def get_flag(self, name):
        return self.flags.get(name)

    def set_flag(self, name, value: dict):
        with self.lock:
            self.flags[name] = value
            self.save()

3️⃣ rollout.py

import hashlib

class RolloutEngine:
    def __init__(self, store):
        self.store = store

    def evaluate_flag(self, name, user_id=None):
        flag = self.store.get_flag(name)
        if not flag:
            return None
        if not flag.get("enabled", False):
            return False

        if not user_id:
            return flag["enabled"]

        bucket = int(hashlib.sha256(user_id.encode()).hexdigest(), 16) % 100
        return bucket < flag.get("rollout", 100)

4️⃣ flags.json (Example)

{
  "some_feature": {
    "enabled": true,
    "rollout": 50
  }
}

🚀 How to Run

1. Install dependencies:

   pip install fastapi uvicorn
   

2. Run the server:

   uvicorn main:app --reload
   

3. Example API calls:

   • GET /flags/new_ui?user_id=123

   • POST /flags/new_ui?enabled=true&rollout_percentage=25

💡 Recap

✅ You built a complete feature flag service
✅ Supports on/off, percentage rollout, and user targeting
✅ Uses clean architecture for scale
✅ Easy to integrate with any backend

✉️ Subscribe for more builds like this

📬 Join hundreds of devs mastering software engineering by building, breaking, and shipping real software.