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

1from datetime import UTC, datetime 

2from enum import StrEnum 

3 

4from beanie import Document 

5from pydantic import Field 

6from pymongo import IndexModel 

7 

8 

9class Destination(StrEnum): 

10 LINKEDIN = "linkedin" 

11 X = "x" 

12 MIXPANEL = "mixpanel" 

13 META = "meta" 

14 GOOGLE_ADS = "google_ads" 

15 

16 

17class DispatchStatus(StrEnum): 

18 PENDING = "pending" 

19 PROCESSING = "processing" 

20 SENT = "sent" 

21 SKIPPED = "skipped" 

22 FAILED = "failed" 

23 

24 

25class ConversionDispatchDb(Document): 

26 """One purchase conversion owed to one destination for one order. 

27 

28 Payloads are rebuilt from the order at delivery time rather than stored, so 

29 this collection never holds anything but ids, status and errors. 

30 """ 

31 

32 provider: str 

33 order_id: str 

34 destination: Destination 

35 status: DispatchStatus = DispatchStatus.PENDING 

36 skip_reason: str | None = None 

37 attempts: int = 0 

38 last_error: str | None = None 

39 next_attempt_at: datetime | None = None 

40 sent_at: datetime | None = None 

41 claimed_at: datetime | None = None 

42 claim_token: str | None = None 

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

44 

45 class Settings: 

46 name = "conversion_dispatches" 

47 indexes = [ 

48 IndexModel( 

49 [("provider", 1), ("order_id", 1), ("destination", 1)], 

50 unique=True, 

51 name="uniq_order_destination", 

52 ), 

53 # Both claim predicates start with the destinations this replica can deliver, so the 

54 # pending backlog of an unconfigured destination is never scanned to find due work. 

55 IndexModel( 

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

57 name="dispatch_queue", 

58 ), 

59 IndexModel([("destination", 1), ("status", 1), ("claimed_at", 1)], name="dispatch_leases"), 

60 ]