Coverage for polar/integrations/loops/client.py: 83%
46 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from typing import TypedDict, Unpack 1a
3import httpx 1a
4import structlog 1a
6from polar.config import settings 1a
7from polar.enums import AccountType 1a
8from polar.logging import Logger 1a
10log: Logger = structlog.get_logger() 1a
13class Properties(TypedDict, total=False): 1a
14 # Loops default properties
15 firstName: str 1a
16 lastName: str 1a
17 notes: str 1a
18 source: str 1a
19 userGroup: str 1a
20 userId: str 1a
21 subscribed: bool 1a
22 createdAt: str 1a
24 # Polar custom properties
25 signupIntent: str 1a
26 emailLogin: bool 1a
27 githubLogin: bool 1a
28 googleLogin: bool 1a
30 organizationCreated: bool 1a
31 organizationSlug: str 1a
32 organizationCount: int 1a
34 productCreated: bool 1a
35 userPatCreated: bool 1a
36 storefrontEnabled: bool 1a
37 webhooksCreated: bool 1a
39 accountType: AccountType 1a
41 # Issue Funding
42 githubOrgInstalled: bool 1a
43 githubIssueBadged: bool 1a
46class LoopsClient: 1a
47 def __init__(self, api_key: str | None) -> None: 1a
48 self.client = httpx.AsyncClient( 1a
49 base_url="https://app.loops.so/api/v1",
50 headers={"Authorization": f"Bearer {api_key}"},
51 # Set a MockTransport if API key is None
52 # Basically, we disable Loops request.
53 transport=(
54 httpx.MockTransport(lambda _: httpx.Response(200))
55 if api_key is None
56 else None
57 ),
58 )
60 async def update_contact( 1a
61 self, email: str, id: str, **properties: Unpack[Properties]
62 ) -> None:
63 log.debug("loops.contact.update", email=email, id=id, **properties)
65 response = await self.client.post(
66 "/contacts/update", json={"email": email, "userId": id, **properties}
67 )
68 self._handle_response(response)
70 async def send_event( 1a
71 self,
72 email: str,
73 event_name: str,
74 event_properties: dict[str, str | int | bool] | None = None,
75 **contact_properties: Unpack[Properties],
76 ) -> None:
77 log.debug(
78 "loops.events.send",
79 email=email,
80 event_name=event_name,
81 event_properties=event_properties,
82 **contact_properties,
83 )
85 response = await self.client.post(
86 "/events/send",
87 json={
88 "email": email,
89 "eventName": event_name,
90 "eventProperties": event_properties or {},
91 **contact_properties,
92 },
93 )
94 self._handle_response(response)
96 def _handle_response(self, response: httpx.Response) -> httpx.Response: 1a
97 response.raise_for_status()
98 return response
101client = LoopsClient(settings.LOOPS_API_KEY) 1a
103__all__ = ["client", "Properties"] 1a