Coverage for polar/posthog.py: 56%

1from __future__ import annotations 

2 

3from typing import Any, Literal 

4 

5from posthog import Posthog 

6 

7from polar.auth.models import AuthSubject, Subject, is_organization, is_user 

8from polar.config import settings 

9from polar.models import Organization, User 

10 

11ORGANIZATION_EVENT_DISTINCT_ID = "organization_event" 

12 

13EventCategory = Literal[ 

14 "benefits", 

15 "subscriptions", 

16 "user", 

17 "organizations", 

18 "issues", 

19] 

20EventNoun = str 

21EventVerb = Literal[ 

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] 

42 

43 

44def _build_event_key(category: EventCategory, noun: EventNoun, verb: EventVerb) -> str: 

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}" 

49 

50 

51class Service: 

52 client: Posthog | None = None 

53 

54 def configure(self) -> None: 

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 

58 

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 

63 

64 def has_feature_flag(self, auth_subject: AuthSubject[Subject], flag: str) -> bool: 

65 if not self.client: 

66 return True 

67 

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 ) 

75 

76 return False 

77 

78 def capture( 

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: 86 ↛ 89line 86 didn't jump to line 89 because the condition on line 86 was always true

87 return 

88 

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 ) 

98 

99 def auth_subject_event( 

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): 107 ↛ 109line 107 didn't jump to line 109 because the condition on line 107 was always true

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) 

115 

116 def anonymous_event( 

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 ) 

132 

133 def user_event( 

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 ) 

150 

151 def organization_event( 

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 ) 

170 

171 def identify(self, user: User) -> None: 

172 if not self.client: 

173 return 

174 

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 ) 

182 

183 def user_login( 

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}) 

188 

189 def user_signup( 

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}) 

194 

195 def _get_common_properties(self) -> dict[str, Any]: 

196 return { 

197 "_environment": settings.ENV, 

198 } 

199 

200 def _get_user_properties(self, user: User) -> dict[str, Any]: 

201 user_data = {"email": user.email, "verified": user.email_verified} 

202 

203 signup = {} 

204 signup_attribution = user.signup_attribution 

205 if signup_attribution: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true

206 for key, value in signup_attribution.items(): 

207 signup[f"signup_{key}"] = value 

208 

209 user_data.update(signup) 

210 return user_data 

211 

212 

213posthog = Service() 

214 

215 

216def configure_posthog() -> None: 

217 posthog.configure()