Coverage for benefits / core / analytics.py: 72%
99 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-01 15:39 +0000
1"""
2The core application: analytics implementation.
3"""
5import itertools
6import json
7import logging
8import re
9import time
10import uuid
12import requests
13from django.conf import settings
15from benefits import VERSION
17from . import models, session
19logger = logging.getLogger(__name__)
22class Event:
23 """Base analytics event of a given type, including attributes from request's session."""
25 _counter = itertools.count()
26 _domain_re = re.compile(r"^(?:https?:\/\/)?(?:[^@\n]+@)?(?:www\.)?([^:\/\n?]+)", re.IGNORECASE)
28 def __init__(self, request, event_type, enrollment_method=models.EnrollmentMethods.DIGITAL, agency=None, **kwargs):
29 self.app_version = VERSION
30 # device_id is generated based on the user_id, and both are set explicitly (per session)
31 self.device_id = session.did(request)
32 self.event_properties = {}
33 self.event_type = str(event_type).lower()
34 self.insert_id = str(uuid.uuid4())
35 self.language = session.language(request)
36 # Amplitude tracks sessions using the start time as the session_id
37 self.session_id = session.start(request)
38 self.time = int(time.time() * 1000)
39 # Although Amplitude advises *against* setting user_id for anonymous users, here a value is set on anonymous
40 # users anyway, as the users never sign-in and become de-anonymized to this app / Amplitude.
41 self.user_id = session.uid(request)
42 self.user_properties = {}
43 self.__dict__.update(kwargs)
45 # Use agency argument if present, otherwise look for one in the session
46 agency = agency or session.agency(request)
47 agency_name = str(agency) if agency else None
49 flow = session.flow(request)
50 verifier_name = flow.eligibility_verifier if flow else None
52 self.update_event_properties(
53 path=request.path,
54 transit_agency=agency_name,
55 eligibility_verifier=verifier_name,
56 enrollment_method=enrollment_method,
57 )
59 uagent = request.headers.get("user-agent")
61 ref = request.headers.get("referer")
62 match = Event._domain_re.match(ref) if ref else None
63 refdom = match.group(1) if match else None
65 self.update_user_properties(
66 referrer=ref,
67 referring_domain=refdom,
68 user_agent=uagent,
69 transit_agency=agency_name,
70 eligibility_verifier=verifier_name,
71 enrollment_method=enrollment_method,
72 )
74 if flow:
75 self.update_enrollment_flows(flow)
77 # event is initialized, consume next counter
78 self.event_id = next(Event._counter)
80 def __str__(self):
81 return json.dumps(self.__dict__)
83 def update_event_properties(self, **kwargs):
84 """Merge kwargs into the self.event_properties dict."""
85 self.event_properties.update(kwargs)
87 def update_user_properties(self, **kwargs):
88 """Merge kwargs into the self.user_properties dict."""
89 self.user_properties.update(kwargs)
91 def update_enrollment_flows(self, flow: models.EnrollmentFlow):
92 enrollment_flows = [flow.system_name]
93 self.update_event_properties(
94 enrollment_flows=enrollment_flows,
95 )
96 self.update_user_properties(
97 enrollment_flows=enrollment_flows,
98 )
101class ViewedPageEvent(Event):
102 """Analytics event representing a single page view."""
104 def __init__(self, request):
105 super().__init__(request, "viewed page")
106 # Add UTM codes
107 utm_campaign = request.GET.get("utm_campaign")
108 utm_source = request.GET.get("utm_source")
109 utm_medium = request.GET.get("utm_medium")
110 utm_content = request.GET.get("utm_content")
111 utm_id = request.GET.get("utm_id")
112 self.update_event_properties(
113 utm_campaign=utm_campaign, utm_source=utm_source, utm_medium=utm_medium, utm_content=utm_content, utm_id=utm_id
114 )
115 self.update_user_properties(
116 utm_campaign=utm_campaign, utm_source=utm_source, utm_medium=utm_medium, utm_content=utm_content, utm_id=utm_id
117 )
120class ChangedLanguageEvent(Event):
121 """Analytics event representing a change in the app's language."""
123 def __init__(self, request, new_lang):
124 super().__init__(request, "changed language")
125 self.update_event_properties(language=new_lang)
128class Client:
129 """Analytics API client"""
131 def __init__(self, api_key):
132 self.api_key = api_key
133 self.headers = {"Accept": "*/*", "Content-type": "application/json"}
134 self.url = "https://api2.amplitude.com/2/httpapi"
135 logger.debug(f"Initialize Client for {self.url}")
137 def _payload(self, events):
138 if not isinstance(events, list):
139 events = [events]
140 return {"api_key": self.api_key, "events": [e.__dict__ for e in events]}
142 def send(self, event):
143 """Send an analytics event."""
144 if not isinstance(event, Event): 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true
145 raise ValueError("event must be an Event instance")
147 if not self.api_key: 147 ↛ 151line 147 didn't jump to line 151 because the condition on line 147 was always true
148 logger.warning(f"api_key is not configured, cannot send event: {event}")
149 return
151 try:
152 payload = self._payload(event)
153 logger.debug(f"Sending event payload: {payload}")
155 r = requests.post(
156 self.url,
157 headers=self.headers,
158 json=payload,
159 timeout=settings.REQUESTS_TIMEOUT,
160 )
161 if r.status_code == 200:
162 logger.debug(f"Event sent successfully: {r.json()}")
163 elif r.status_code == 400:
164 logger.error(f"Event request was invalid: {r.json()}")
165 elif r.status_code == 413:
166 logger.error(f"Event payload was too large: {r.json()}")
167 elif r.status_code == 429:
168 logger.error(f"Event contained too many requests for some users: {r.json()}")
169 else:
170 logger.error(f"Failed to send event: {r.json()}")
172 except Exception:
173 logger.error(f"Failed to send event: {event}")
176client = Client(settings.ANALYTICS_KEY)
179def send_event(event):
180 """Send an analytics event."""
181 if isinstance(event, Event): 181 ↛ 184line 181 didn't jump to line 184 because the condition on line 181 was always true
182 client.send(event)
183 else:
184 raise ValueError("event must be an Event instance")