How to add feature flags and experiments to a Django backend

# Keep AMPLITUDE_API_KEY from the product analytics tutorial.
export AMPLITUDE_DEPLOYMENT_KEY="your-server-deployment-key"
export AMPLITUDE_SERVER_ZONE="US"

uv add "amplitude-experiment>=1.13,<2"

# Keep analytics and authentication from the product analytics tutorial.
import logging
import os
from threading import Lock
from amplitude_experiment import Experiment, RemoteEvaluationConfig, User
from amplitude_experiment.remote.fetch_options import FetchOptions
from amplitude_experiment.server_zone import ServerZone

FLAG_KEY = "report-summary"
_client = None
_lock = Lock()
log = logging.getLogger(__name__)


def get_experiment():
    global _client
    with _lock:
        if _client is None:
            zone = ServerZone.EU if os.environ.get("AMPLITUDE_SERVER_ZONE", "US") == "EU" else ServerZone.US
            _client = Experiment.initialize_remote(
                os.environ["AMPLITUDE_DEPLOYMENT_KEY"],
                RemoteEvaluationConfig(
                    server_zone=zone,
                    fetch_timeout_millis=500,
                    fetch_retries=0,
                    connection_pool_max_size=8,
                    fetch_pool_acquire_timeout_millis=100,
                ),
            )
        return _client


def choose_report_layout(user_id, account_type):
    # This experiment is for signed-in users. Auth determines eligibility.
    if user_id is None:
        return "control"
    client = get_experiment()
    try:
        variants = client.fetch_v2(
            User(user_id=user_id, user_properties={"account_type": account_type}),
            FetchOptions(tracksExposure=True, flagKeys=[FLAG_KEY]),
        ) or {}
        variant = variants.get(FLAG_KEY)
        if variant is not None and variant.value in {"control", "treatment"}:
            return variant.value
    except Exception:
        # Availability wins for this cosmetic layout. Keep authorization separate.
        log.warning("Experiment evaluation failed; serving the baseline report")
    return "control"


def close_experiment():
    # Call only at worker exit, after requests have drained.
    if _client is not None:
        _client.close()

from amplitude import BaseEvent
from .analytics import get_client


def record_report_export(user_id, device_id, framework):
    if not isinstance(device_id, str) or not 5 <= len(device_id) <= 1024:
        device_id = None
    if user_id is None and device_id is None:
        return False
    get_client().track(BaseEvent(
        event_type="Report Exported",
        user_id=user_id,
        device_id=device_id,
        event_properties={"report_type": "usage", "days": 7, "framework": framework},
    ))
    return True

# Keep routes and decorators from /build/django-product-analytics.
from django.http import HttpResponse, JsonResponse
from django.middleware.csrf import get_token
from django.views.decorators.http import require_http_methods, require_POST
from .auth import get_current_user
from .analytics import record_report_preview
from .experiments import choose_report_layout
from .report_events import record_report_export


@require_http_methods(["GET", "POST"])
def report_preview(request):
    if request.method == "GET":
        return JsonResponse({"csrfToken": get_token(request)})
    user = get_current_user(request)
    user_id = f"user-{user.pk}" if user is not None else None
    account_type = "staff" if user is not None and user.is_staff else "member" if user is not None else None
    # Evaluate only where the selected experience is about to be used.
    layout = choose_report_layout(user_id, account_type)
    preview = {"report_type": "usage", "days": 7, "layout": layout}
    if layout == "treatment":
        preview["summary"] = "Your last seven days of usage are ready to export."
    record_report_preview(user_id, request.headers.get("X-Amplitude-Device-Id"), "django", account_type)
    return JsonResponse(preview)


@require_POST
def report_export(request):
    user = get_current_user(request)
    user_id = f"user-{user.pk}" if user is not None else None
    # No flag fetch here: exporting is the outcome, not another exposure.
    response = HttpResponse("report_type,days\nusage,7\n", content_type="text/csv")
    response["Content-Disposition"] = 'attachment; filename="usage.csv"'
    record_report_export(user_id, request.headers.get("X-Amplitude-Device-Id"), "django")
    return response

# Add export beside the existing preview; keep all other URL patterns.
from django.urls import path
from .views import report_preview, report_export

urlpatterns = [
    path("reports/preview", report_preview),
    path("reports/export", report_export),
]

# Keep your existing authenticated cookies in cookies.txt.
ORIGIN="http://127.0.0.1:8000"
CSRF_TOKEN=$(curl -sS -b cookies.txt -c cookies.txt "$ORIGIN/api/reports/preview" |
  python3 -c 'import json,sys; print(json.load(sys.stdin)["csrfToken"])')
curl --fail-with-body -b cookies.txt -X POST "$ORIGIN/api/reports/preview"   -H "X-CSRFToken: $CSRF_TOKEN"   -H "X-Amplitude-Device-Id: existing-browser-device-id"

curl --fail-with-body -b cookies.txt -X POST "$ORIGIN/api/reports/export" -H "X-CSRFToken: $CSRF_TOKEN" -H "X-Amplitude-Device-Id: existing-browser-device-id" -o usage.csv

# -- metrics/experiments.py --
# Keep analytics and authentication from the product analytics tutorial.
import logging
import os
from threading import Lock
from amplitude_experiment import Experiment, RemoteEvaluationConfig, User
from amplitude_experiment.remote.fetch_options import FetchOptions
from amplitude_experiment.server_zone import ServerZone

FLAG_KEY = "report-summary"
_client = None
_lock = Lock()
log = logging.getLogger(__name__)


def get_experiment():
    global _client
    with _lock:
        if _client is None:
            zone = ServerZone.EU if os.environ.get("AMPLITUDE_SERVER_ZONE", "US") == "EU" else ServerZone.US
            _client = Experiment.initialize_remote(
                os.environ["AMPLITUDE_DEPLOYMENT_KEY"],
                RemoteEvaluationConfig(
                    server_zone=zone,
                    fetch_timeout_millis=500,
                    fetch_retries=0,
                    connection_pool_max_size=8,
                    fetch_pool_acquire_timeout_millis=100,
                ),
            )
        return _client


def choose_report_layout(user_id, account_type):
    # This experiment is for signed-in users. Auth determines eligibility.
    if user_id is None:
        return "control"
    client = get_experiment()
    try:
        variants = client.fetch_v2(
            User(user_id=user_id, user_properties={"account_type": account_type}),
            FetchOptions(tracksExposure=True, flagKeys=[FLAG_KEY]),
        ) or {}
        variant = variants.get(FLAG_KEY)
        if variant is not None and variant.value in {"control", "treatment"}:
            return variant.value
    except Exception:
        # Availability wins for this cosmetic layout. Keep authorization separate.
        log.warning("Experiment evaluation failed; serving the baseline report")
    return "control"


def close_experiment():
    # Call only at worker exit, after requests have drained.
    if _client is not None:
        _client.close()

# -- metrics/report_events.py --
from amplitude import BaseEvent
from .analytics import get_client


def record_report_export(user_id, device_id, framework):
    if not isinstance(device_id, str) or not 5 <= len(device_id) <= 1024:
        device_id = None
    if user_id is None and device_id is None:
        return False
    get_client().track(BaseEvent(
        event_type="Report Exported",
        user_id=user_id,
        device_id=device_id,
        event_properties={"report_type": "usage", "days": 7, "framework": framework},
    ))
    return True

# -- metrics/views.py --
# Keep routes and decorators from /build/django-product-analytics.
from django.http import HttpResponse, JsonResponse
from django.middleware.csrf import get_token
from django.views.decorators.http import require_http_methods, require_POST
from .auth import get_current_user
from .analytics import record_report_preview
from .experiments import choose_report_layout
from .report_events import record_report_export


@require_http_methods(["GET", "POST"])
def report_preview(request):
    if request.method == "GET":
        return JsonResponse({"csrfToken": get_token(request)})
    user = get_current_user(request)
    user_id = f"user-{user.pk}" if user is not None else None
    account_type = "staff" if user is not None and user.is_staff else "member" if user is not None else None
    # Evaluate only where the selected experience is about to be used.
    layout = choose_report_layout(user_id, account_type)
    preview = {"report_type": "usage", "days": 7, "layout": layout}
    if layout == "treatment":
        preview["summary"] = "Your last seven days of usage are ready to export."
    record_report_preview(user_id, request.headers.get("X-Amplitude-Device-Id"), "django", account_type)
    return JsonResponse(preview)


@require_POST
def report_export(request):
    user = get_current_user(request)
    user_id = f"user-{user.pk}" if user is not None else None
    # No flag fetch here: exporting is the outcome, not another exposure.
    response = HttpResponse("report_type,days\nusage,7\n", content_type="text/csv")
    response["Content-Disposition"] = 'attachment; filename="usage.csv"'
    record_report_export(user_id, request.headers.get("X-Amplitude-Device-Id"), "django")
    return response

# -- metrics/urls.py --
# Add export beside the existing preview; keep all other URL patterns.
from django.urls import path
from .views import report_preview, report_export

urlpatterns = [
    path("reports/preview", report_preview),
    path("reports/export", report_export),
]

Extend the existing Django report example using https://amplitude.com/build/django-feature-flags.
Keep metrics/analytics.py and metrics/auth.py from /build/django-product-analytics unchanged, including real verified authentication and worker shutdown flushing. Merge route and lifespan changes; preserve unrelated code and authorization/CSRF controls.
Use amplitude-experiment>=1.13,<2 with one remote client per worker. Read AMPLITUDE_DEPLOYMENT_KEY only on the server; it must belong to the analytics project. Honor AMPLITUDE_SERVER_ZONE for both SDKs.
Use the tested FetchOptions import from amplitude_experiment.remote.fetch_options, with tracksExposure=True and flagKeys=["report-summary"]. Supply a new User with the canonical authenticated ID and trusted account_type per request.
Evaluate only at POST /api/reports/preview. Serve control or treatment; only treatment adds summary. Anonymous identity, missing/unknown variants, and evaluation errors use the baseline without inventing exposures. Keep authentication separate from the flag.
Add POST /api/reports/export to return CSV and send Report Exported after preparing the response. Do not fetch a flag or manually send exposure on export. Keep the same user ID as evaluation and product analytics.
Keep both POSTs protected by Django CSRF and merge URL patterns.
Verify both variants, property targeting context, scoped exposure requests, conversion identity, timeout/failure fallback, concurrency, and shutdown delivery. Confirm remote exposures and experiment analysis separately in Amplitude. Report every file changed.