Feature Flags

Control feature rollout without deploying code.

Why Feature Flags?

• Gradual rollout — Release to 1%, then 10%, then 100%
• Quick rollback — Disable features without deploying
• A/B testing — Test variations with real users
• Trunk-based development — Merge incomplete features

Simple Implementation

Configuration-Based Flags

# feature_flags.py
from enum import Enum
from typing import Optional
from pydantic_settings import BaseSettings, SettingsConfigDict

class Environment(str, Enum):
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"

class FeatureFlags(BaseSettings):
    environment: Environment = Environment.DEVELOPMENT

    # Feature definitions
    NEW_CHECKOUT: bool = False
    DARK_MODE: bool = True
    AI_RECOMMENDATIONS: bool = False

    model_config = SettingsConfigDict(env_prefix="FF_")  # FF_NEW_CHECKOUT=true

flags = FeatureFlags()

def is_enabled(feature: str) -> bool:
    """Check if a feature is enabled."""
    return getattr(flags, feature.upper(), False)

# Usage
if is_enabled("NEW_CHECKOUT"):
    return new_checkout_flow()
else:
    return legacy_checkout_flow()

Environment-Specific Flags

FEATURE_FLAGS = {
    "new_checkout": {
        "development": True,
        "staging": True,
        "production": False,
    },
    "dark_mode": {
        "development": True,
        "staging": True,
        "production": True,
    },
    "ai_recommendations": {
        "development": True,
        "staging": True,
        "production": {
            "enabled": True,
            "percentage": 10,  # 10% of users
        },
    },
}

def is_enabled(feature: str, user_id: Optional[str] = None) -> bool:
    config = FEATURE_FLAGS.get(feature, {})
    env_config = config.get(settings.environment)

    if isinstance(env_config, bool):
        return env_config

    if isinstance(env_config, dict):
        if not env_config.get("enabled", False):
            return False

        # Percentage-based rollout
        if "percentage" in env_config and user_id:
            return hash(f"{feature}:{user_id}") % 100 < env_config["percentage"]

        return env_config.get("enabled", False)

    return False

Database-Backed Flags

Model

from sqlalchemy import Column, String, Boolean, JSON, DateTime
from sqlalchemy.orm import declarative_base
from datetime import datetime

Base = declarative_base()

class FeatureFlag(Base):
    __tablename__ = "feature_flags"

    id = Column(String, primary_key=True)
    name = Column(String, unique=True, nullable=False)
    enabled = Column(Boolean, default=False)
    config = Column(JSON, default={})
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, onupdate=datetime.utcnow)

# Example config:
# {
#     "percentage": 50,
#     "user_ids": ["user-123", "user-456"],
#     "segments": ["beta_testers"],
#     "start_date": "2024-01-01",
#     "end_date": "2024-02-01",
# }

Service

from functools import lru_cache
from datetime import datetime, date
import hashlib

class FeatureFlagService:
    def __init__(self, db):
        self.db = db
        self._cache: dict = {}
        self._cache_ttl = 60  # seconds

    async def is_enabled(
        self,
        feature: str,
        user_id: Optional[str] = None,
        user_segments: Optional[list[str]] = None,
    ) -> bool:
        flag = await self._get_flag(feature)
        if not flag or not flag.enabled:
            return False

        config = flag.config or {}

        # Check date range
        if "start_date" in config:
            if date.today() < date.fromisoformat(config["start_date"]):
                return False

        if "end_date" in config:
            if date.today() > date.fromisoformat(config["end_date"]):
                return False

        # Check specific user IDs
        if "user_ids" in config:
            if user_id in config["user_ids"]:
                return True

        # Check segments
        if "segments" in config and user_segments:
            if any(s in config["segments"] for s in user_segments):
                return True

        # Percentage rollout
        if "percentage" in config and user_id:
            hash_value = int(hashlib.md5(f"{feature}:{user_id}".encode()).hexdigest(), 16)
            return (hash_value % 100) < config["percentage"]

        return flag.enabled

    async def _get_flag(self, feature: str) -> Optional[FeatureFlag]:
        # Check cache
        cached = self._cache.get(feature)
        if cached and cached["expires"] > datetime.utcnow().timestamp():
            return cached["flag"]

        # Fetch from database
        flag = await self.db.query(FeatureFlag).filter(
            FeatureFlag.name == feature
        ).first()

        # Update cache
        self._cache[feature] = {
            "flag": flag,
            "expires": datetime.utcnow().timestamp() + self._cache_ttl,
        }

        return flag

    async def set_flag(
        self,
        feature: str,
        enabled: bool,
        config: Optional[dict] = None,
    ):
        flag = await self._get_flag(feature)

        if flag:
            flag.enabled = enabled
            if config:
                flag.config = config
        else:
            flag = FeatureFlag(
                id=str(uuid4()),
                name=feature,
                enabled=enabled,
                config=config or {},
            )
            self.db.add(flag)

        await self.db.commit()

        # Invalidate cache
        self._cache.pop(feature, None)

Frontend Integration

React Context

import { createContext, useContext, useEffect, useState } from 'react';

interface FeatureFlags {
  [key: string]: boolean;
}

const FeatureFlagContext = createContext<FeatureFlags>({});

export function FeatureFlagProvider({ children }: { children: React.ReactNode }) {
  const [flags, setFlags] = useState<FeatureFlags>({});

  useEffect(() => {
    // Fetch flags from API
    fetch('/api/feature-flags')
      .then(res => res.json())
      .then(setFlags);
  }, []);

  return (
    <FeatureFlagContext.Provider value={flags}>
      {children}
    </FeatureFlagContext.Provider>
  );
}

export function useFeatureFlag(feature: string): boolean {
  const flags = useContext(FeatureFlagContext);
  return flags[feature] ?? false;
}

// Usage
function CheckoutPage() {
  const newCheckoutEnabled = useFeatureFlag('new_checkout');

  if (newCheckoutEnabled) {
    return <NewCheckout />;
  }

  return <LegacyCheckout />;
}

Feature Component

interface FeatureProps {
  name: string;
  children: React.ReactNode;
  fallback?: React.ReactNode;
}

export function Feature({ name, children, fallback = null }: FeatureProps) {
  const isEnabled = useFeatureFlag(name);
  return isEnabled ? <>{children}</> : <>{fallback}</>;
}

// Usage
function App() {
  return (
    <div>
      <Feature name="dark_mode">
        <DarkModeToggle />
      </Feature>

      <Feature name="new_dashboard" fallback={<LegacyDashboard />}>
        <NewDashboard />
      </Feature>
    </div>
  );
}

A/B Testing

Variant Assignment

import hashlib
from dataclasses import dataclass
from typing import Optional

@dataclass
class Experiment:
    name: str
    variants: list[str]
    weights: list[int]  # e.g., [50, 50] for 50/50 split

def get_variant(experiment: Experiment, user_id: str) -> str:
    """Deterministic variant assignment based on user ID."""
    hash_value = int(
        hashlib.md5(f"{experiment.name}:{user_id}".encode()).hexdigest(),
        16
    )

    # Calculate cumulative weights
    total_weight = sum(experiment.weights)
    bucket = hash_value % total_weight

    cumulative = 0
    for variant, weight in zip(experiment.variants, experiment.weights):
        cumulative += weight
        if bucket < cumulative:
            return variant

    return experiment.variants[-1]

# Usage
experiment = Experiment(
    name="checkout_flow",
    variants=["control", "variant_a", "variant_b"],
    weights=[34, 33, 33],  # ~33% each
)

variant = get_variant(experiment, user.id)  # Always same for user

Tracking

async def track_experiment_exposure(
    experiment: str,
    variant: str,
    user_id: str,
):
    await analytics.track("experiment_exposure", {
        "experiment": experiment,
        "variant": variant,
        "user_id": user_id,
        "timestamp": datetime.utcnow().isoformat(),
    })

async def track_conversion(
    experiment: str,
    variant: str,
    user_id: str,
    event: str,
    value: Optional[float] = None,
):
    await analytics.track("experiment_conversion", {
        "experiment": experiment,
        "variant": variant,
        "user_id": user_id,
        "event": event,
        "value": value,
        "timestamp": datetime.utcnow().isoformat(),
    })

Third-Party Services

LaunchDarkly

import ldclient
from ldclient import Context

ldclient.set_config(ldclient.Config(settings.launchdarkly_sdk_key))
client = ldclient.get()

def is_enabled(feature: str, user: User) -> bool:
    context = Context.builder(user.id) \
        .set("email", user.email) \
        .set("plan", user.plan) \
        .build()

    return client.variation(feature, context, False)

PostHog

from posthog import Posthog

posthog = Posthog(settings.posthog_api_key)

def is_enabled(feature: str, user_id: str) -> bool:
    return posthog.feature_enabled(feature, user_id)

def get_variant(feature: str, user_id: str) -> str:
    return posthog.get_feature_flag(feature, user_id)

Admin API

from fastapi import APIRouter, Depends, HTTPException

router = APIRouter(prefix="/admin/flags")

@router.get("/")
async def list_flags(
    admin: Admin = Depends(require_admin),
    service: FeatureFlagService = Depends(),
):
    return await service.list_all()

@router.post("/{feature}")
async def update_flag(
    feature: str,
    enabled: bool,
    config: Optional[dict] = None,
    admin: Admin = Depends(require_admin),
    service: FeatureFlagService = Depends(),
):
    await service.set_flag(feature, enabled, config)

    # Audit log
    await audit_log.record(
        action="feature_flag_updated",
        actor=admin.id,
        resource=feature,
        details={"enabled": enabled, "config": config},
    )

    return {"status": "updated"}

@router.delete("/{feature}")
async def delete_flag(
    feature: str,
    admin: Admin = Depends(require_admin),
    service: FeatureFlagService = Depends(),
):
    await service.delete_flag(feature)
    return {"status": "deleted"}

Best Practices

Practice	Benefit
Default to off	Safe deploys
Use consistent hashing	Same user gets same variant
Set expiration dates	Clean up old flags
Log flag evaluations	Debug issues
Test both paths	Ensure both work
Document flags	Team awareness
Clean up old flags	Reduce complexity
Audit changes	Accountability

Flag Lifecycle

┌─────────────┐
│   Create    │  Feature in development
└──────┬──────┘
       │
┌──────┴──────┐
│   Staged    │  Enabled in staging only
└──────┬──────┘
       │
┌──────┴──────┐
│   Rollout   │  Gradual % increase
└──────┬──────┘
       │
┌──────┴──────┐
│   Released  │  100% enabled
└──────┬──────┘
       │
┌──────┴──────┐
│   Cleanup   │  Remove flag, code cleanup
└─────────────┘