Coverage for atlas_ecommerce/purchases/shopify.py: 100%

1from datetime import UTC, datetime 

2 

3from pydantic import ValidationError 

4 

5from atlas_ecommerce.attribution import OrderAttribution, from_note_attributes 

6from atlas_ecommerce.models import OrderDb, ShopifyPurchaseOrder 

7from atlas_ecommerce.purchases import PurchaseEvent 

8 

9PROVIDER = "shopify" 

10PAID_TOPIC = "orders/paid" 

11PAID_STATUSES = frozenset({"paid", "partially_refunded"}) 

12 

13 

14def unreportable(order: OrderDb, problem: str) -> PurchaseEvent: 

15 return PurchaseEvent( 

16 provider=PROVIDER, 

17 order_id=str(order.shopify_order_id), 

18 happened_at=datetime.now(UTC), 

19 attribution=OrderAttribution(), 

20 paid=False, 

21 problem=problem, 

22 source_ref=order, 

23 ) 

24 

25 

26def to_purchase(order: OrderDb) -> PurchaseEvent: 

27 if not order.raw: 

28 return unreportable(order, "order has no payload") 

29 try: 

30 shopify_order = ShopifyPurchaseOrder.model_validate(order.raw) 

31 except ValidationError as exc: 

32 return unreportable(order, f"order payload unparseable: {exc.error_count()} validation error(s)") 

33 if shopify_order.happened_at is None: 

34 return unreportable(order, "order carries no timestamp") 

35 return PurchaseEvent( 

36 provider=PROVIDER, 

37 order_id=str(shopify_order.id), 

38 happened_at=shopify_order.happened_at, 

39 attribution=from_note_attributes((a.name, a.value) for a in shopify_order.note_attributes), 

40 value=shopify_order.total_price, 

41 currency=shopify_order.currency, 

42 quantity=shopify_order.quantity, 

43 variant_id=shopify_order.first_variant_id, 

44 order_number=shopify_order.order_number, 

45 paid=shopify_order.financial_status in PAID_STATUSES, 

46 source_ref=order, 

47 ) 

48 

49 

50class ShopifyPurchases: 

51 provider = PROVIDER 

52 

53 async def unplanned(self, limit: int) -> list[PurchaseEvent]: 

54 # Deliberately not the provisioning topic setting: a purchase is an orders/paid delivery and 

55 # nothing else. Under any other subscription the queue stays empty instead of stamping 

56 # pending orders as planned. 

57 orders: list[OrderDb] = ( 

58 await OrderDb.find({"conversions_enqueued_at": None, "topic": PAID_TOPIC}, projection_model=OrderDb) 

59 .limit(limit) 

60 .to_list() 

61 ) 

62 return [to_purchase(order) for order in orders] 

63 

64 async def mark_planned(self, purchase: PurchaseEvent) -> None: 

65 # Field-level: the provisioning worker may hold the same document and save its own snapshot. 

66 await purchase.source_ref.set({"conversions_enqueued_at": datetime.now(UTC)}) 

67 

68 async def load(self, order_id: str) -> PurchaseEvent | None: 

69 order = await OrderDb.find_one({"shopify_order_id": int(order_id)}) 

70 return to_purchase(order) if order is not None else None