Coverage for polar/models/webhook_endpoint.py: 97%

1from enum import StrEnum 

2from typing import TYPE_CHECKING, Literal 

3from uuid import UUID 

4 

5from sqlalchemy import Boolean, ForeignKey, String, Uuid 

6from sqlalchemy.dialects.postgresql import JSONB 

7from sqlalchemy.orm import Mapped, declared_attr, mapped_column, relationship 

8 

9from polar.kit.db.models.base import RecordModel 

10 

11if TYPE_CHECKING: 11 ↛ 12line 11 didn't jump to line 12 because the condition on line 11 was never true

12 from .organization import Organization 

13 

14 

15class WebhookEventType(StrEnum): 

16 checkout_created = "checkout.created" 

17 checkout_updated = "checkout.updated" 

18 customer_created = "customer.created" 

19 customer_updated = "customer.updated" 

20 customer_deleted = "customer.deleted" 

21 customer_state_changed = "customer.state_changed" 

22 customer_seat_assigned = "customer_seat.assigned" 

23 customer_seat_claimed = "customer_seat.claimed" 

24 customer_seat_revoked = "customer_seat.revoked" 

25 order_created = "order.created" 

26 order_updated = "order.updated" 

27 order_paid = "order.paid" 

28 order_refunded = "order.refunded" 

29 subscription_created = "subscription.created" 

30 subscription_updated = "subscription.updated" 

31 subscription_active = "subscription.active" 

32 subscription_canceled = "subscription.canceled" 

33 subscription_uncanceled = "subscription.uncanceled" 

34 subscription_revoked = "subscription.revoked" 

35 refund_created = "refund.created" 

36 refund_updated = "refund.updated" 

37 product_created = "product.created" 

38 product_updated = "product.updated" 

39 benefit_created = "benefit.created" 

40 benefit_updated = "benefit.updated" 

41 benefit_grant_created = "benefit_grant.created" 

42 benefit_grant_cycled = "benefit_grant.cycled" 

43 benefit_grant_updated = "benefit_grant.updated" 

44 benefit_grant_revoked = "benefit_grant.revoked" 

45 organization_updated = "organization.updated" 

46 

47 

48CustomerWebhookEventType = Literal[ 

49 WebhookEventType.customer_created, 

50 WebhookEventType.customer_updated, 

51 WebhookEventType.customer_deleted, 

52 WebhookEventType.customer_state_changed, 

53] 

54 

55 

56class WebhookFormat(StrEnum): 

57 raw = "raw" 

58 discord = "discord" 

59 slack = "slack" 

60 

61 

62class WebhookEndpoint(RecordModel): 

63 __tablename__ = "webhook_endpoints" 

64 

65 url: Mapped[str] = mapped_column(String, nullable=False) 

66 format: Mapped[WebhookFormat] = mapped_column(String, nullable=False) 

67 secret: Mapped[str] = mapped_column(String, nullable=False) 

68 events: Mapped[list[WebhookEventType]] = mapped_column( 

69 JSONB, nullable=False, default=[] 

70 ) 

71 enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) 

72 

73 organization_id: Mapped[UUID] = mapped_column( 

74 Uuid, 

75 ForeignKey("organizations.id", ondelete="CASCADE"), 

76 nullable=False, 

77 index=True, 

78 ) 

79 

80 @declared_attr 

81 def organization(cls) -> Mapped["Organization"]: 

82 return relationship("Organization", lazy="raise")