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

1from datetime import UTC, datetime, timedelta 

2 

3import httpx 

4 

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

6from atlas_ecommerce.purchases import PurchaseEvent 

7 

8CONVERSIONS_URL = "https://ads-api.x.com/12/measurement/conversions/{pixel_id}" 

9# Static Conversions API token minted in Events Manager; replaces OAuth 1.0a for this endpoint. 

10TOKEN_HEADER = "X-Pixel-Token" 

11MAX_EVENT_AGE = timedelta(days=90) 

12 

13 

14def to_iso_millis(moment: datetime) -> str: 

15 utc = moment.astimezone(UTC) 

16 return utc.strftime("%Y-%m-%dT%H:%M:%S.") + f"{utc.microsecond // 1000:03d}Z" 

17 

18 

19def purchase_conversion(purchase: PurchaseEvent, event_id: str) -> dict: 

20 return { 

21 "conversions": [ 

22 { 

23 "conversion_time": to_iso_millis(purchase.happened_at), 

24 "event_id": event_id, 

25 "identifiers": [{"twclid": purchase.attribution.click_ids["twclid"]}], 

26 "conversion_id": purchase.order_id, 

27 "value": purchase.value, 

28 "price_currency": purchase.currency, 

29 "number_items": purchase.quantity, 

30 } 

31 ] 

32 } 

33 

34 

35def unprocessed_conversions(response: httpx.Response, expected: int) -> str | None: 

36 """X answers 2xx and only reports how many events it kept; a shortfall is the sole sign of a rejection.""" 

37 try: 

38 data = response.json().get("data") or {} 

39 except (ValueError, AttributeError): 

40 return f"unreadable response body: {response.text[:200]}" 

41 processed = data.get("conversions_processed") 

42 if not isinstance(processed, int) or processed < expected: 

43 return f"conversions_processed={processed!r} of {expected} sent (debug_id {data.get('debug_id')})" 

44 return None 

45 

46 

47class XConversionsClient: 

48 def __init__(self, *, pixel_id: str, pixel_token: str, http: httpx.AsyncClient | None = None) -> None: 

49 self._url = CONVERSIONS_URL.format(pixel_id=pixel_id) 

50 self._headers = {TOKEN_HEADER: pixel_token, "Content-Type": "application/json"} 

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

52 

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

54 try: 

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

56 except httpx.TransportError as exc: 

57 raise TransientDeliveryError(f"X Ads API unreachable: {exc}") from exc 

58 raise_for_response(response) 

59 problem = unprocessed_conversions(response, len(payload.get("conversions") or [])) 

60 if problem: 

61 raise PermanentDeliveryError(f"X accepted the request but did not process the conversion: {problem}") 

62 

63 async def close(self) -> None: 

64 await self._http.aclose()