Coverage for polar/integrations/loops/service.py: 34%
52 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1from typing import Unpack 1a
3from polar.enums import AccountType 1a
4from polar.models import User 1a
5from polar.postgres import AsyncSession 1a
6from polar.user_organization.service import ( 1a
7 user_organization as user_organization_service,
8)
9from polar.worker import enqueue_job 1a
11from .client import Properties 1a
14class Loops: 1a
15 #####################################################################
16 # CREATOR-ONLY EVENTS: We can skip `is_creator` check on all below
17 #####################################################################
19 async def user_organization_added(self, session: AsyncSession, user: User) -> None: 1a
20 user_organizations = await user_organization_service.list_by_user_id(
21 session, user.id
22 )
23 await self.enqueue_event(
24 user,
25 event="Organization Created",
26 properties={
27 "organizationCreated": True,
28 # Always use the first organization.
29 # Loops contacts are 1:1 with users vs. organizations so we can
30 # only keep reference to one (main) organization to link to etc.
31 "organizationSlug": user_organizations[0].organization.slug,
32 "organizationCount": len(user_organizations),
33 },
34 )
36 async def user_created_product(self, user: User) -> None: 1a
37 await self.enqueue_event(
38 user,
39 event="Product Created",
40 properties={
41 "productCreated": True,
42 },
43 )
45 async def user_installed_github_organization(self, user: User) -> None: 1a
46 await self.enqueue_event(
47 user,
48 event="GitHub Organization Installed",
49 properties={
50 "githubOrgInstalled": True,
51 },
52 )
54 async def user_badged_github_issue(self, user: User) -> None: 1a
55 await self.enqueue_event(
56 user,
57 event="GitHub Issue Badged",
58 properties={
59 "githubIssueBadged": True,
60 },
61 )
63 #####################################################################
64 # USER EVENTS: We have to check `is_creator`
65 #####################################################################
67 async def user_signup( 1a
68 self,
69 user: User,
70 **properties: Unpack[Properties],
71 ) -> None:
72 # Only create contacts for creators on signup.
73 # Others can be created later on upon first creator events (flywheel)
74 if not user.had_creator_signup_intent:
75 return
77 await self.enqueue_event(
78 user,
79 event="User Signed Up",
80 properties={
81 # Set login method in `properties` to override defaults (False)
82 "emailLogin": False,
83 "githubLogin": False,
84 "googleLogin": False,
85 "organizationCreated": False,
86 "organizationCount": 0,
87 "productCreated": False,
88 "userPatCreated": False,
89 "storefrontEnabled": False,
90 "githubOrgInstalled": False,
91 "githubIssueBadged": False,
92 **properties,
93 },
94 )
96 async def user_update( 1a
97 self, session: AsyncSession, user: User, **properties: Unpack[Properties]
98 ) -> None:
99 is_creator = await self.is_creator(session, user)
100 if not is_creator:
101 return
103 properties = self.get_updated_user_properties(user, properties)
104 enqueue_job("loops.update_contact", user.email, str(user.id), **properties)
106 async def user_created_personal_access_token( 1a
107 self, session: AsyncSession, user: User
108 ) -> None:
109 is_creator = await self.is_creator(session, user)
110 if not is_creator:
111 return
113 await self.enqueue_event(
114 user,
115 event="User PAT Created",
116 properties={
117 "userPatCreated": True,
118 },
119 )
121 async def user_created_account( 1a
122 self, session: AsyncSession, user: User, accountType: AccountType
123 ) -> None:
124 is_creator = await self.is_creator(session, user)
125 if not is_creator:
126 return
128 await self.enqueue_event(
129 user,
130 event="User Finance Account Created",
131 properties={
132 "accountType": accountType,
133 },
134 )
136 #####################################################################
137 # HELPERS
138 #####################################################################
140 async def is_creator(self, session: AsyncSession, user: User) -> bool: 1a
141 if user.had_creator_signup_intent:
142 return True
144 user_organization_count = (
145 await user_organization_service.get_user_organization_count(
146 session, user.id
147 )
148 )
149 return user_organization_count > 0
151 def get_updated_user_properties( 1a
152 self, user: User, properties: Properties
153 ) -> Properties:
154 signup_intent = user.signup_attribution.get("intent")
155 updated: Properties = {
156 "userId": str(user.id),
157 "userGroup": "creator",
158 "signupIntent": signup_intent or "",
159 }
160 updated.update(properties)
161 return updated
163 async def enqueue_event( 1a
164 self, user: User, *, event: str, properties: Properties
165 ) -> None:
166 properties = self.get_updated_user_properties(user, properties)
167 enqueue_job("loops.send_event", user.email, event, **properties)
170loops = Loops() 1a
172__all__ = ["loops"] 1a