Coverage for polar/integrations/loops/client.py: 67%

1from typing import TypedDict, Unpack 

2 

3import httpx 

4import structlog 

5 

6from polar.config import settings 

7from polar.enums import AccountType 

8from polar.logging import Logger 

9 

10log: Logger = structlog.get_logger() 

11 

12 

13class Properties(TypedDict, total=False): 

14 # Loops default properties 

15 firstName: str 

16 lastName: str 

17 notes: str 

18 source: str 

19 userGroup: str 

20 userId: str 

21 subscribed: bool 

22 createdAt: str 

23 

24 # Polar custom properties 

25 signupIntent: str 

26 emailLogin: bool 

27 githubLogin: bool 

28 googleLogin: bool 

29 

30 organizationCreated: bool 

31 organizationSlug: str 

32 organizationCount: int 

33 

34 productCreated: bool 

35 userPatCreated: bool 

36 storefrontEnabled: bool 

37 webhooksCreated: bool 

38 

39 accountType: AccountType 

40 

41 # Issue Funding 

42 githubOrgInstalled: bool 

43 githubIssueBadged: bool 

44 

45 

46class LoopsClient: 

47 def __init__(self, api_key: str | None) -> None: 

48 self.client = httpx.AsyncClient( 

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 ) 

59 

60 async def update_contact( 

61 self, email: str, id: str, **properties: Unpack[Properties] 

62 ) -> None: 

63 log.debug("loops.contact.update", email=email, id=id, **properties) 

64 

65 response = await self.client.post( 

66 "/contacts/update", json={"email": email, "userId": id, **properties} 

67 ) 

68 self._handle_response(response) 

69 

70 async def send_event( 

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 ) 

84 

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) 

95 

96 def _handle_response(self, response: httpx.Response) -> httpx.Response: 

97 response.raise_for_status() 

98 return response 

99 

100 

101client = LoopsClient(settings.LOOPS_API_KEY) 

102 

103__all__ = ["client", "Properties"]