Coverage for polar/checkout/eventstream.py: 72%

1from enum import StrEnum 

2from typing import Literal, TypedDict, overload 

3 

4from polar.eventstream.service import publish 

5from polar.models.checkout import CheckoutStatus 

6 

7 

8class CheckoutEvent(StrEnum): 

9 updated = "checkout.updated" 

10 order_created = "checkout.order_created" 

11 subscription_created = "checkout.subscription_created" 

12 webhook_event_delivered = "checkout.webhook_event_delivered" 

13 

14 

15class CheckoutEventUpdatedPayload(TypedDict): 

16 status: CheckoutStatus 

17 

18 

19class CheckoutEventWebhookEventDeliveredPayload(TypedDict): 

20 status: CheckoutStatus 

21 

22 

23@overload 

24async def publish_checkout_event( 24 ↛ anywhereline 24 didn't jump anywhere: it always raised an exception.

25 client_secret: str, 

26 event: Literal[CheckoutEvent.updated], 

27 payload: CheckoutEventUpdatedPayload, 

28) -> None: ... 

29 

30 

31@overload 

32async def publish_checkout_event( 32 ↛ anywhereline 32 didn't jump anywhere: it always raised an exception.

33 client_secret: str, 

34 event: Literal[CheckoutEvent.order_created], 

35) -> None: ... 

36 

37 

38@overload 

39async def publish_checkout_event( 39 ↛ anywhereline 39 didn't jump anywhere: it always raised an exception.

40 client_secret: str, 

41 event: Literal[CheckoutEvent.subscription_created], 

42) -> None: ... 

43 

44 

45@overload 

46async def publish_checkout_event( 46 ↛ anywhereline 46 didn't jump anywhere: it always raised an exception.

47 client_secret: str, 

48 event: Literal[CheckoutEvent.webhook_event_delivered], 

49 payload: CheckoutEventWebhookEventDeliveredPayload, 

50) -> None: ... 

51 

52 

53async def publish_checkout_event( 

54 client_secret: str, 

55 event: CheckoutEvent, 

56 payload: CheckoutEventUpdatedPayload 

57 | CheckoutEventWebhookEventDeliveredPayload 

58 | None = None, 

59) -> None: 

60 return await publish( 

61 event, {**payload} if payload else {}, checkout_client_secret=client_secret 

62 )