Coverage for atlas_ecommerce/clients/meta_capi.py: 92%

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 

8EVENTS_URL = "https://graph.facebook.com/{version}/{pixel_id}/events" 

9MAX_EVENT_AGE = timedelta(days=7) 

10# Graph API signals throttling with HTTP 400 and one of these codes, not with 429. 

11THROTTLE_CODES = frozenset({1, 2, 4, 17, 32, 341, 613, *range(80000, 80015)}) 

12# Session/token errors: only a new system-user token fixes them, the event is fine. 

13AUTH_CODES = frozenset({102, 190}) 

14 

15 

16def fbc_for(purchase: PurchaseEvent) -> str | None: 

17 """The pixel's own _fbc cookie when the site captured it; otherwise Meta's documented synthesis 

18 fb.<subdomain index>.<creation ms>.<fbclid>, stamped with the best observation time we have.""" 

19 attribution = purchase.attribution 

20 if attribution.fbc: 

21 return attribution.fbc 

22 fbclid = attribution.click_ids.get("fbclid") 

23 if not fbclid: 

24 return None 

25 return f"fb.1.{int(purchase.happened_at.timestamp() * 1000)}.{fbclid}" 

26 

27 

28def graph_error(response: httpx.Response) -> dict: 

29 try: 

30 return response.json().get("error") or {} 

31 except ValueError: 

32 return {} 

33 

34 

35def throttled(response: httpx.Response) -> bool: 

36 if response.status_code != 400: 

37 return False 

38 error = graph_error(response) 

39 return bool(error.get("is_transient")) or error.get("code") in THROTTLE_CODES 

40 

41 

42def unauthenticated(response: httpx.Response) -> bool: 

43 return response.status_code in (401, 403) or graph_error(response).get("code") in AUTH_CODES 

44 

45 

46def purchase_event(purchase: PurchaseEvent, test_event_code: str | None = None) -> dict: 

47 attribution = purchase.attribution 

48 user_data: dict = {"client_user_agent": attribution.user_agent} 

49 if attribution.fbp: 

50 user_data["fbp"] = attribution.fbp 

51 fbc = fbc_for(purchase) 

52 if fbc: 

53 user_data["fbc"] = fbc 

54 custom_data: dict = { 

55 "currency": purchase.currency, 

56 "value": float(purchase.value) if purchase.value else 0.0, 

57 "order_id": purchase.order_id, 

58 "num_items": purchase.quantity, 

59 "content_type": "product", 

60 } 

61 if purchase.variant_id is not None: 

62 custom_data["content_ids"] = [str(purchase.variant_id)] 

63 payload: dict = { 

64 "data": [ 

65 { 

66 "event_name": "Purchase", 

67 "event_time": int(purchase.happened_at.timestamp()), 

68 "event_id": purchase.order_id, 

69 "action_source": "website", 

70 "user_data": user_data, 

71 "custom_data": custom_data, 

72 } 

73 ] 

74 } 

75 # Routes the event to Events Manager's Test events tab instead of reporting; verification only. 

76 if test_event_code: 

77 payload["test_event_code"] = test_event_code 

78 return payload 

79 

80 

81class MetaConversionsClient: 

82 def __init__( 

83 self, *, pixel_id: str, access_token: str, api_version: str, http: httpx.AsyncClient | None = None 

84 ) -> None: 

85 self._url = EVENTS_URL.format(version=api_version, pixel_id=pixel_id) 

86 # A request parameter or a bearer header, never a field of the event document; the header keeps it out of URLs. 

87 self._headers = {"Authorization": f"Bearer {access_token}"} 

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

89 

90 async def send(self, payload: dict) -> None: 

91 try: 

92 response = await self._http.post(self._url, json=payload, headers=self._headers) 

93 except httpx.TransportError as exc: 

94 raise TransientDeliveryError(f"Meta Graph API unreachable: {exc}") from exc 

95 if throttled(response): 

96 raise TransientDeliveryError(f"Meta throttled the request: {response.text[:300]}") 

97 if not response.is_success and unauthenticated(response): 

98 raise TransientDeliveryError(f"Meta refused the access token; rotate it: {response.text[:300]}") 

99 raise_for_response(response) 

100 body = response.json() 

101 if not body.get("events_received"): 

102 raise PermanentDeliveryError(f"Meta accepted the request but recorded no event: {body}") 

103 

104 async def close(self) -> None: 

105 await self._http.aclose()