Coverage for atlas_ecommerce/clients/mixpanel.py: 89%

1from datetime import timedelta 

2 

3import httpx 

4 

5from atlas_ecommerce.clients.delivery import PermanentDeliveryError, TransientDeliveryError, raise_for_response 

6from atlas_ecommerce.purchases import PurchaseEvent 

7 

8# ip=0: the request comes from our server, whose location must not be stamped on the buyer's event. 

9TRACK_URL = "https://api.mixpanel.com/track?ip=0&verbose=1" 

10EVENT_NAME = "purchase" 

11# /track only ingests the last five days; anything older needs /import and a project secret. 

12MAX_EVENT_AGE = timedelta(days=5) 

13 

14 

15def purchase_event(purchase: PurchaseEvent, token: str) -> dict: 

16 attribution = purchase.attribution 

17 properties: dict = { 

18 "token": token, 

19 "distinct_id": attribution.mp_distinct_id, 

20 "$insert_id": purchase.order_id, 

21 "time": int(purchase.happened_at.timestamp()), 

22 "order_id": purchase.order_id, 

23 "order_number": purchase.order_number, 

24 "value": float(purchase.value) if purchase.value else None, 

25 "currency": purchase.currency, 

26 "variant_id": purchase.variant_id, 

27 "quantity": purchase.quantity, 

28 "source": f"{purchase.provider}_checkout", 

29 **attribution.mixpanel_properties, 

30 } 

31 if attribution.mp_device_id: 

32 properties["$device_id"] = attribution.mp_device_id 

33 return {"event": EVENT_NAME, "properties": properties} 

34 

35 

36class MixpanelClient: 

37 def __init__(self, *, http: httpx.AsyncClient | None = None) -> None: 

38 self._http = http or httpx.AsyncClient(timeout=15.0) 

39 

40 async def send(self, event: dict) -> None: 

41 try: 

42 response = await self._http.post( 

43 TRACK_URL, 

44 json=[event], 

45 headers={"Accept": "text/plain", "Content-Type": "application/json"}, 

46 ) 

47 except httpx.TransportError as exc: 

48 raise TransientDeliveryError(f"Mixpanel unreachable: {exc}") from exc 

49 raise_for_response(response) 

50 body = response.json() 

51 if body.get("status") != 1: 

52 raise PermanentDeliveryError(f"Mixpanel refused the event: {body.get('error')}") 

53 

54 async def close(self) -> None: 

55 await self._http.aclose()