Coverage for atlas_ecommerce/models/order.py: 100%

1"""Order records mirrored from Shopify webhook deliveries.""" 

2 

3from datetime import UTC, datetime 

4from enum import StrEnum 

5from typing import Any 

6 

7from beanie import Document 

8from pydantic import BaseModel, Field 

9from pymongo import IndexModel 

10 

11 

12class OrderStatus(StrEnum): 

13 RECEIVED = "received" 

14 PROVISIONED = "provisioned" 

15 FAILED = "failed" 

16 

17 

18class OrderDb(Document): 

19 """One Shopify order, plus the account provisioning that followed it. 

20 

21 The raw payload is kept verbatim: Shopify adds fields over time and the 

22 parsed projection below is deliberately narrow. 

23 """ 

24 

25 shopify_order_id: int 

26 webhook_id: str | None = None 

27 topic: str | None = None 

28 shop_domain: str | None = None 

29 

30 email: str | None = None 

31 first_name: str | None = None 

32 last_name: str | None = None 

33 order_number: int | None = None 

34 financial_status: str | None = None 

35 

36 raw: dict[str, Any] = Field(default_factory=dict) 

37 

38 status: OrderStatus = OrderStatus.RECEIVED 

39 stytch_user_id: str | None = None 

40 attempts: int = 0 

41 last_error: str | None = None 

42 

43 received_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) 

44 provisioned_at: datetime | None = None 

45 # When a failed order becomes eligible again. None means immediately. 

46 next_attempt_at: datetime | None = None 

47 conversions_enqueued_at: datetime | None = None 

48 

49 class Settings: 

50 name = "orders" 

51 indexes = [ 

52 IndexModel("shopify_order_id", unique=True, name="uniq_shopify_order_id"), 

53 # Sparse: deliveries replayed by hand carry no webhook id. 

54 IndexModel("webhook_id", unique=True, sparse=True, name="uniq_webhook_id"), 

55 IndexModel( 

56 [("status", 1), ("attempts", 1), ("next_attempt_at", 1)], 

57 name="provision_queue", 

58 ), 

59 # Not sparse: the conversion worker polls for null/missing, and pre-existing orders have no field. 

60 IndexModel([("topic", 1), ("conversions_enqueued_at", 1)], name="conversions_enqueue_queue"), 

61 ] 

62 

63 

64class ShopifyCustomer(BaseModel): 

65 first_name: str | None = None 

66 last_name: str | None = None 

67 

68 

69class ShopifyNoteAttribute(BaseModel): 

70 name: str 

71 value: Any = None 

72 

73 

74class ShopifyLineItem(BaseModel): 

75 quantity: int = 1 

76 variant_id: int | None = None 

77 

78 

79class ShopifyOrder(BaseModel): 

80 """What intake and provisioning need. Everything else stays in ``raw``.""" 

81 

82 id: int 

83 email: str | None = None 

84 order_number: int | None = None 

85 financial_status: str | None = None 

86 customer: ShopifyCustomer | None = None 

87 

88 @property 

89 def contact_email(self) -> str | None: 

90 """Shopify sends "" rather than null for an absent email.""" 

91 return self.email.strip() or None if self.email else None 

92 

93 

94class ShopifyPurchaseOrder(ShopifyOrder): 

95 """The extra fields conversion delivery reads, parsed from ``raw`` only when a purchase is built.""" 

96 

97 note_attributes: list[ShopifyNoteAttribute] = Field(default_factory=list) 

98 total_price: str | None = None 

99 currency: str | None = None 

100 processed_at: datetime | None = None 

101 created_at: datetime | None = None 

102 line_items: list[ShopifyLineItem] = Field(default_factory=list) 

103 

104 @property 

105 def happened_at(self) -> datetime | None: 

106 moment = self.processed_at or self.created_at 

107 if moment is not None and moment.tzinfo is None: 

108 return moment.replace(tzinfo=UTC) 

109 return moment 

110 

111 @property 

112 def quantity(self) -> int: 

113 return sum(item.quantity for item in self.line_items) or 1 

114 

115 @property 

116 def first_variant_id(self) -> int | None: 

117 return next((item.variant_id for item in self.line_items if item.variant_id is not None), None)