FastAPI Feature Flags

Introduction

Feature flags (also known as feature toggles or feature switches) are a powerful technique in modern software development that allows developers to modify system behavior without changing code. They enable teams to turn features on or off at runtime, gradually roll out features to users, conduct A/B testing, and deploy code that isn't fully ready yet.

In this tutorial, we'll explore how to implement feature flags in FastAPI applications, covering both simple approaches and more sophisticated solutions.

What Are Feature Flags?

Feature flags are conditional statements in your code that determine whether a feature is available to a particular user or in a specific environment. They help you:

Safely deploy features that are still in progress
Gradually roll out features to a subset of users
Conduct A/B testing to compare different implementations
Quickly disable problematic features without redeploying
Provide different experiences to different user segments

Basic Implementation of Feature Flags in FastAPI

Let's start with a simple implementation of feature flags in a FastAPI application.

1. Using Environment Variables

The simplest approach is to use environment variables to control features:

python
import os
from fastapi import FastAPI

app = FastAPI()

# Check if feature is enabled via environment variable
ENABLE_NEW_UI = os.getenv("ENABLE_NEW_UI", "false").lower() == "true"

@app.get("/")
async def root():
    if ENABLE_NEW_UI:
        return {"message": "Welcome to the new UI!"}
    else:
        return {"message": "Welcome to the classic UI!"}

You can run this application with the feature enabled like this:

bash
ENABLE_NEW_UI=true uvicorn main:app

2. Using Configuration Files

For more complex applications, you might want to use configuration files:

python
import yaml
from fastapi import FastAPI

app = FastAPI()

# Load feature flags from a YAML file
def load_feature_flags():
    with open("feature_flags.yaml", "r") as file:
        return yaml.safe_load(file)

feature_flags = load_feature_flags()

@app.get("/")
async def root():
    if feature_flags.get("enable_new_ui", False):
        return {"message": "Welcome to the new UI!"}
    else:
        return {"message": "Welcome to the classic UI!"}

Example feature_flags.yaml file:

yaml
enable_new_ui: false
enable_new_payment_processing: true
enable_beta_features: false

Advanced Feature Flag Implementation

For more realistic applications, we need a more sophisticated approach to feature flags.

Creating a Feature Flag Service

Let's create a dedicated feature flag service:

python
from fastapi import FastAPI, Depends, HTTPException, Request
from pydantic import BaseModel
import yaml
from typing import Dict, Optional, List, Callable
import random

app = FastAPI()

class FeatureFlag(BaseModel):
    name: str
    enabled: bool
    # Percentage of users who should see this feature (0-100)
    rollout_percentage: Optional[int] = 100
    # List of user IDs that should always see this feature
    user_ids: Optional[List[str]] = []
    # List of user groups that should see this feature
    groups: Optional[List[str]] = []

class FeatureFlagService:
    def __init__(self, flags_file: str = "feature_flags.yaml"):
        self.flags_file = flags_file
        self.reload_flags()
    
    def reload_flags(self):
        with open(self.flags_file, "r") as file:
            data = yaml.safe_load(file)
            self.flags = {name: FeatureFlag(name=name, **config) 
                          for name, config in data.items()}
    
    def is_enabled(self, flag_name: str, user_id: Optional[str] = None, 
                  user_groups: Optional[List[str]] = None) -> bool:
        """Check if a feature flag is enabled for a specific user"""
        if flag_name not in self.flags:
            return False
            
        flag = self.flags[flag_name]
        
        # If the flag is disabled, return False immediately
        if not flag.enabled:
            return False
            
        # If user ID is in the whitelist, enable the feature
        if user_id and flag.user_ids and user_id in flag.user_ids:
            return True
            
        # Check if user belongs to an enabled group
        if user_groups and flag.groups:
            if any(group in flag.groups for group in user_groups):
                return True
        
        # Apply percentage rollout if no other conditions match
        if flag.rollout_percentage < 100:
            # Use user_id to ensure consistent experience
            if user_id:
                # Deterministic random based on user_id and flag name
                import hashlib
                seed = f"{user_id}:{flag_name}"
                hash_value = int(hashlib.md5(seed.encode()).hexdigest(), 16)
                return (hash_value % 100) < flag.rollout_percentage
            else:
                # Random percentage for anonymous users
                return random.randint(1, 100) <= flag.rollout_percentage
                
        return True

# Create a singleton instance of the service
feature_flag_service = FeatureFlagService()

# Dependency to get the feature flag service
def get_feature_flag_service():
    return feature_flag_service

Now, let's create some routes that use our feature flag service:

python
@app.get("/api/features")
async def get_features(
    feature_flags: FeatureFlagService = Depends(get_feature_flag_service)
):
    """Return all feature flags (admin only in a real app)"""
    return feature_flags.flags

@app.get("/")
async def root(
    request: Request,
    feature_flags: FeatureFlagService = Depends(get_feature_flag_service)
):
    # In a real app, you'd get the user ID from the authentication system
    user_id = request.headers.get("X-User-ID")
    user_groups = request.headers.get("X-User-Groups", "").split(",") if request.headers.get("X-User-Groups") else []
    
    # Check if the new UI feature is enabled for this user
    new_ui_enabled = feature_flags.is_enabled("enable_new_ui", user_id, user_groups)
    
    if new_ui_enabled:
        return {"message": "Welcome to the new UI!"}
    else:
        return {"message": "Welcome to the classic UI!"}

@app.get("/payment")
async def payment(
    request: Request,
    feature_flags: FeatureFlagService = Depends(get_feature_flag_service)
):
    user_id = request.headers.get("X-User-ID")
    user_groups = request.headers.get("X-User-Groups", "").split(",") if request.headers.get("X-User-Groups") else []
    
    if feature_flags.is_enabled("new_payment_system", user_id, user_groups):
        return {"message": "Using the new payment system!"}
    else:
        return {"message": "Using the classic payment system!"}

Example feature_flags.yaml for the advanced implementation:

yaml
enable_new_ui:
  enabled: true
  rollout_percentage: 25
  user_ids:
    - "admin123"
    - "tester456"
  groups:
    - "beta_testers"

new_payment_system:
  enabled: true
  rollout_percentage: 10
  groups:
    - "internal_users"

experimental_api:
  enabled: false

Creating Feature Flag Middleware

To automatically check feature flags for specific routes, we can create custom middleware:

python
from fastapi import FastAPI, Depends, HTTPException, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware

class FeatureFlagMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, feature_flag_service: FeatureFlagService):
        super().__init__(app)
        self.feature_flag_service = feature_flag_service
        # Define which endpoints require which feature flags
        self.route_flags = {
            "/api/v2/": "api_v2",
            "/experimental/": "experimental_features"
        }

    async def dispatch(self, request: Request, call_next):
        # Check if any route prefix matches
        for route_prefix, flag_name in self.route_flags.items():
            if request.url.path.startswith(route_prefix):
                user_id = request.headers.get("X-User-ID")
                user_groups = request.headers.get("X-User-Groups", "").split(",") if request.headers.get("X-User-Groups") else []
                
                if not self.feature_flag_service.is_enabled(flag_name, user_id, user_groups):
                    return Response(
                        status_code=404,
                        content="Not Found",
                        media_type="text/plain"
                    )
        
        # If no feature flag restrictions or all passed, continue with the request
        response = await call_next(request)
        return response

# Add middleware to the application
app.add_middleware(
    FeatureFlagMiddleware, 
    feature_flag_service=feature_flag_service
)

# Now all requests to /api/v2/* will only work if the "api_v2" flag is enabled
@app.get("/api/v2/users")
async def get_users_v2():
    return {"users": ["User1", "User2"], "version": "v2"}

Feature Flag Decorator

For more fine-grained control, we can create a decorator to protect specific endpoints:

python
from functools import wraps
from fastapi import HTTPException, Request, Depends

def require_feature(flag_name: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(
            request: Request,
            feature_flags: FeatureFlagService = Depends(get_feature_flag_service),
            *args, **kwargs
        ):
            user_id = request.headers.get("X-User-ID")
            user_groups = request.headers.get("X-User-Groups", "").split(",") if request.headers.get("X-User-Groups") else []
            
            if not feature_flags.is_enabled(flag_name, user_id, user_groups):
                raise HTTPException(status_code=404, detail="Feature not available")
            
            return await func(request=request, *args, **kwargs)
        return wrapper
    return decorator

# Use the decorator to protect endpoints
@app.get("/beta-feature")
@require_feature("beta_features")
async def beta_feature(request: Request):
    return {"message": "You have access to the beta feature!"}

Real-world Applications

1. Gradual Rollout of a New UI

Imagine you're updating your application's UI. You can use feature flags to gradually roll it out:

python
@app.get("/dashboard")
async def dashboard(
    request: Request,
    feature_flags: FeatureFlagService = Depends(get_feature_flag_service),
    templates: Jinja2Templates = Depends(get_templates)
):
    user_id = request.cookies.get("user_id")
    user_groups = get_user_groups(user_id)  # Custom function to get user groups
    
    if feature_flags.is_enabled("new_dashboard", user_id, user_groups):
        return templates.TemplateResponse("new_dashboard.html", {"request": request})
    else:
        return templates.TemplateResponse("dashboard.html", {"request": request})

2. A/B Testing for Performance Optimization

You can implement A/B testing to compare different algorithms:

python
@app.get("/search")
async def search(
    query: str,
    request: Request,
    feature_flags: FeatureFlagService = Depends(get_feature_flag_service)
):
    user_id = request.cookies.get("user_id")
    
    # Use different search algorithms for different users
    if feature_flags.is_enabled("new_search_algorithm", user_id):
        results = new_search_algorithm(query)
        # Log that the new algorithm was used
        log_search_metrics(user_id, "new_algorithm", query, results)
    else:
        results = old_search_algorithm(query)
        # Log that the old algorithm was used
        log_search_metrics(user_id, "old_algorithm", query, results)
        
    return {"results": results}

3. Feature Gates for Subscription Tiers

Use feature flags to control access to premium features:

python
@app.get("/advanced-analytics")
async def advanced_analytics(
    request: Request,
    feature_flags: FeatureFlagService = Depends(get_feature_flag_service),
    user_service: UserService = Depends(get_user_service)
):
    user_id = request.cookies.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Authentication required")
    
    user = user_service.get_user(user_id)
    user_tier = user.subscription_tier  # e.g., "free", "premium", "enterprise"
    
    # Check if this user's tier should have access to this feature
    if not feature_flags.is_enabled(f"analytics_{user_tier}"):
        raise HTTPException(status_code=403, detail="Upgrade your subscription to access this feature")
        
    # User has access to the feature
    return {"analytics_data": generate_advanced_analytics(user_id)}

Using External Feature Flag Services

For production applications, consider using dedicated feature flag services that provide additional capabilities like:

Web UI for managing flags
User targeting rules
A/B testing analytics
Audit logs

Here's how to integrate with a third-party service like LaunchDarkly:

python
import ldclient
from ldclient.config import Config

# Initialize LaunchDarkly client
ldclient.set_config(Config("YOUR_SDK_KEY"))

class LaunchDarklyFeatureService:
    def is_enabled(self, flag_name: str, user_id: str, user_attributes: dict = None) -> bool:
        if not user_id:
            # Anonymous user
            context = {"key": "anonymous"}
        else:
            context = {
                "key": user_id,
                "kind": "user",
                **(user_attributes or {})
            }
            
        return ldclient.get().variation(flag_name, context, False)

# Usage
@app.get("/premium-feature")
async def premium_feature(
    request: Request,
    ld_client: LaunchDarklyFeatureService = Depends(get_launchdarkly_service)
):
    user_id = request.cookies.get("user_id", "anonymous")
    user_data = get_user_data(user_id)  # Custom function to get user data
    
    # Check if the feature is enabled for this user
    if ld_client.is_enabled("premium_feature", user_id, user_data):
        return {"message": "Premium feature is available!"}
    else:
        return {"message": "Premium feature is not available for your account."}

Summary

Feature flags are a powerful technique that allows you to:

Safely deploy code while controlling feature availability
Gradually roll out new features to users
Conduct A/B tests to validate changes
Quickly disable problematic features without rolling back code
Provide different feature sets to different user segments

When implementing feature flags in FastAPI, you can start with simple approaches like environment variables or configuration files. For more complex applications, consider creating a dedicated feature flag service or using third-party services.

The examples in this tutorial showed different approaches to implement feature flags, from basic to advanced, including:

Simple environment-based flags
Configuration file-based flags
A full-featured flag service with user targeting
Route-specific middleware for protecting entire API sections
Decorators for protecting individual endpoints
Integration with external feature flag services

Additional Resources

Exercises

Implement a simple feature flag system using environment variables in a FastAPI application.
Create a feature flag service that loads flags from a JSON or YAML file.
Implement a feature flag that gradually rolls out a new feature to 10% of users.
Create a middleware that protects all routes under /experimental/ with a feature flag.
Add A/B testing to a search endpoint, with logging to compare the performance of two algorithms.

If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)

Introduction​

What Are Feature Flags?​

Basic Implementation of Feature Flags in FastAPI​

1. Using Environment Variables​

2. Using Configuration Files​

Advanced Feature Flag Implementation​

Creating a Feature Flag Service​

Creating Feature Flag Middleware​

Feature Flag Decorator​

Real-world Applications​

1. Gradual Rollout of a New UI​

2. A/B Testing for Performance Optimization​

3. Feature Gates for Subscription Tiers​

Using External Feature Flag Services​

Summary​

Additional Resources​

Exercises​