Coverage for polar/posthog.py: 37%
68 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1from __future__ import annotations 1a
3from typing import Any, Literal 1a
5from posthog import Posthog 1a
7from polar.auth.models import AuthSubject, Subject, is_organization, is_user 1a
8from polar.config import settings 1a
9from polar.models import Organization, User 1a
11ORGANIZATION_EVENT_DISTINCT_ID = "organization_event" 1a
13EventCategory = Literal[ 1a
14 "benefits",
15 "subscriptions",
16 "user",
17 "organizations",
18 "issues",
19]
20EventNoun = str 1a
21EventVerb = Literal[ 1a
22 "click",
23 "submit",
24 "create",
25 "view",
26 "add",
27 "invite",
28 "update",
29 "delete",
30 "remove",
31 "start",
32 "end",
33 "cancel",
34 "fail",
35 "generate",
36 "send",
37 "archive",
38 "done",
39 "open",
40 "close",
41]
44def _build_event_key(category: EventCategory, noun: EventNoun, verb: EventVerb) -> str: 1a
45 # Surface is interesting in the client-side implementation to determine
46 # which product area the customer engaged with. For programmatic & async
47 # operations on the backend we hardcode all of them to `backend`.
48 return f"backend:{category}:{noun}:{verb}"
51class Service: 1a
52 client: Posthog | None = None 1a
54 def configure(self) -> None: 1a
55 if not settings.POSTHOG_PROJECT_API_KEY: 55 ↛ 59line 55 didn't jump to line 59 because the condition on line 55 was always true
56 self.client = None
57 return
59 self.client = Posthog(settings.POSTHOG_PROJECT_API_KEY)
60 self.client.disabled = settings.is_testing()
61 self.client.debug = settings.POSTHOG_DEBUG
62 self.client.feature_enabled
64 def has_feature_flag(self, auth_subject: AuthSubject[Subject], flag: str) -> bool: 1a
65 if not self.client:
66 return True
68 if is_user(auth_subject):
69 return (
70 self.client.feature_enabled(
71 flag, distinct_id=auth_subject.subject.posthog_distinct_id
72 )
73 or False
74 )
76 return False
78 def capture( 1a
79 self,
80 distinct_id: str,
81 event: str,
82 *,
83 properties: dict[str, Any] | None = None,
84 groups: dict[str, Any] | None = None,
85 ) -> None:
86 if not self.client:
87 return
89 self.client.capture(
90 event,
91 distinct_id=distinct_id,
92 groups=groups,
93 properties={
94 **self._get_common_properties(),
95 **(properties or {}),
96 },
97 )
99 def auth_subject_event( 1a
100 self,
101 auth_subject: AuthSubject[Subject],
102 category: EventCategory,
103 noun: EventNoun,
104 verb: EventVerb,
105 properties: dict[str, Any] | None = None,
106 ) -> None:
107 if is_user(auth_subject):
108 self.user_event(auth_subject.subject, category, noun, verb, properties)
109 elif is_organization(auth_subject):
110 self.organization_event(
111 auth_subject.subject, category, noun, verb, properties
112 )
113 else:
114 self.anonymous_event(category, noun, verb, properties)
116 def anonymous_event( 1a
117 self,
118 category: EventCategory,
119 noun: EventNoun,
120 verb: EventVerb,
121 properties: dict[str, Any] | None = None,
122 ) -> None:
123 """Shorthand for one-off anonymous event capture."""
124 self.capture(
125 distinct_id="polar_anonymous",
126 event=_build_event_key(category, noun, verb),
127 properties={
128 **self._get_common_properties(),
129 **(properties or {}),
130 },
131 )
133 def user_event( 1a
134 self,
135 user: User,
136 category: EventCategory,
137 noun: EventNoun,
138 verb: EventVerb,
139 properties: dict[str, Any] | None = None,
140 ) -> None:
141 self.capture(
142 user.posthog_distinct_id,
143 event=_build_event_key(category, noun, verb),
144 properties={
145 **self._get_common_properties(),
146 "$set": self._get_user_properties(user),
147 **(properties or {}),
148 },
149 )
151 def organization_event( 1a
152 self,
153 organization: Organization,
154 category: EventCategory,
155 noun: EventNoun,
156 verb: EventVerb,
157 properties: dict[str, Any] | None = None,
158 ) -> None:
159 self.capture(
160 ORGANIZATION_EVENT_DISTINCT_ID, # Ref: https://posthog.com/docs/product-analytics/group-analytics#advanced-server-side-only-capturing-group-events-without-a-user
161 event=_build_event_key(category, noun, verb),
162 groups={
163 "organization": str(organization.id),
164 },
165 properties={
166 **self._get_common_properties(),
167 **(properties or {}),
168 },
169 )
171 def identify(self, user: User) -> None: 1a
172 if not self.client:
173 return
175 self.client.set(
176 distinct_id=user.posthog_distinct_id,
177 properties={
178 **self._get_common_properties(),
179 **self._get_user_properties(user),
180 },
181 )
183 def user_login( 1a
184 self, user: User, method: Literal["github", "google", "apple", "ml", "code"]
185 ) -> None:
186 self.identify(user)
187 self.user_event(user, "user", "login", "done", {"method": method})
189 def user_signup( 1a
190 self, user: User, method: Literal["github", "google", "apple", "ml", "code"]
191 ) -> None:
192 self.identify(user)
193 self.user_event(user, "user", "signup", "done", {"method": method})
195 def _get_common_properties(self) -> dict[str, Any]: 1a
196 return {
197 "_environment": settings.ENV,
198 }
200 def _get_user_properties(self, user: User) -> dict[str, Any]: 1a
201 user_data = {"email": user.email, "verified": user.email_verified}
203 signup = {}
204 signup_attribution = user.signup_attribution
205 if signup_attribution:
206 for key, value in signup_attribution.items():
207 signup[f"signup_{key}"] = value
209 user_data.update(signup)
210 return user_data
213posthog = Service() 1a
216def configure_posthog() -> None: 1a
217 posthog.configure()