Coverage for polar/models/webhook_event.py: 87%

1from typing import TYPE_CHECKING 2 ctx1ab

2from uuid import UUID 2 ctx1ab

3 

4from sqlalchemy import Boolean, ColumnElement, ForeignKey, Index, Integer, String, Uuid 2 ctx1ab

5from sqlalchemy.ext.hybrid import hybrid_property 2 ctx1ab

6from sqlalchemy.orm import Mapped, declared_attr, mapped_column, relationship 2 ctx1ab

7 

8from polar.kit.db.models.base import RecordModel 2 ctx1ab

9from polar.kit.extensions.sqlalchemy.types import StringEnum 2 ctx1ab

10 

11from .webhook_endpoint import WebhookEventType 2 ctx1ab

12 

13if TYPE_CHECKING: 13 ↛ 14line 13 didn't jump to line 14 because the condition on line 13 was never true2 ctx1ab

14 from .webhook_endpoint import WebhookEndpoint 

15 

16 

17class WebhookEvent(RecordModel): 2 ctx1ab

18 __tablename__ = "webhook_events" 2 ctx1ab

19 __table_args__ = ( 2 ctx1ab

20 Index( 

21 "ix_webhook_events_created_at_non_archived", 

22 "created_at", 

23 postgresql_where="payload IS NOT NULL", 

24 ), 

25 ) 

26 

27 webhook_endpoint_id: Mapped[UUID] = mapped_column( 2 ctx1ab

28 Uuid, 

29 ForeignKey("webhook_endpoints.id", ondelete="CASCADE"), 

30 nullable=False, 

31 index=True, 

32 ) 

33 

34 @declared_attr 2 ctx1ab

35 def webhook_endpoint(cls) -> Mapped["WebhookEndpoint"]: 2 ctx1ab

36 return relationship("WebhookEndpoint", lazy="raise") 2 ctx1ab

37 

38 last_http_code: Mapped[int | None] = mapped_column(Integer, nullable=True) 2 ctx1ab

39 succeeded: Mapped[bool | None] = mapped_column(Boolean, nullable=True) 2 ctx1ab

40 skipped: Mapped[bool] = mapped_column( 2 ctx1ab

41 Boolean, nullable=False, default=False, server_default="false" 

42 ) 

43 type: Mapped[WebhookEventType] = mapped_column( 2 ctx1ab

44 StringEnum(WebhookEventType), nullable=False, index=True 

45 ) 

46 payload: Mapped[str | None] = mapped_column(String, nullable=True) 2 ctx1ab

47 

48 @hybrid_property 2 ctx1ab

49 def is_archived(self) -> bool: 2 ctx1ab

50 return self.payload is None 

51 

52 @is_archived.inplace.expression 2 ctx1ab

53 @classmethod 2 ctx1ab

54 def _is_archived_expression(cls) -> ColumnElement[bool]: 2 ctx1ab

55 return cls.payload.is_(None)