Skip to content

Instantly share code, notes, and snippets.

@sshleifer
Last active October 28, 2022 15:20
Show Gist options
  • Save sshleifer/88cfef883d45b812ed6c7cd5dca25800 to your computer and use it in GitHub Desktop.
Save sshleifer/88cfef883d45b812ed6c7cd5dca25800 to your computer and use it in GitHub Desktop.
from growthbook import GrowthBook
from chat.models import ExperimentInfo
from django.core.cache import cache
from typing import Dict, List, Tuple
def get_growthbook_features() -> Dict[str, Dict]:
import requests
key = "gb_features"
result = cache.get(key)
if result is None:
apiResp = requests.get(
"https://cdn.growthbook.io/api/features/REDACTED"
)
result = apiResp.json()["features"]
cache.set(key, result, timeout=60 * 5)
return result
class GrowthbookMiddleware:
def __init__(self, get_response):
print("init MiddleWare")
self.get_response = get_response
def __call__(self, request):
# Create unique session identifier (works with anon users!)
if request.session.session_key is None:
request.session.create()
print("__call__ MiddleWare")
features: Dict = get_growthbook_features()
#print(f'{features=}')
# User attributes for targeting and experimentation
# Must match https://app.growthbook.io/attributes ?
# Try to have a key for anonymous users
attributes = {
"id": request.user.id,
"sessionId": request.session.session_key,
"idOrSessionId": str(request.user.id) or request.session.session_key,
}
def on_experiment_viewed(experiment, result):
# This never gets hit
print(f"{type(experiment.key)=}, {type(result.variationId)=}")
raise ValueError(f"{type(experiment.key)=}, {type(result.variationId)=}")
# TODO: figure out how often this is called, maybe batch inserts (i.e.,
# accumulate in this function, then do bulk insert after
# self.get_response(request))
obj = {k: v for k, v in attributes.items()}
obj["experimentId"] = experiment.key
obj["variationId"] = result.variationId
ExperimentInfo.objects.get_or_create(**obj)
# Create a GrowthBook instance
request.gb = GrowthBook(
attributes=attributes,
features=features,
trackingCallback=on_experiment_viewed,
)
response = self.get_response(request)
request.gb.destroy() # Cleanup
return response
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment