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

1from typing import TYPE_CHECKING 

2from uuid import UUID 

3 

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

5from sqlalchemy.ext.hybrid import hybrid_property 

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

7 

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

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

10 

11from .webhook_endpoint import WebhookEventType 

12 

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

14 from .webhook_endpoint import WebhookEndpoint 

15 

16 

17class WebhookEvent(RecordModel): 

18 __tablename__ = "webhook_events" 

19 __table_args__ = ( 

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( 

28 Uuid, 

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

30 nullable=False, 

31 index=True, 

32 ) 

33 

34 @declared_attr 

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

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

37 

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

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

40 skipped: Mapped[bool] = mapped_column( 

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

42 ) 

43 type: Mapped[WebhookEventType] = mapped_column( 

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

45 ) 

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

47 

48 @hybrid_property 

49 def is_archived(self) -> bool: 

50 return self.payload is None 

51 

52 @is_archived.inplace.expression 

53 @classmethod 

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

55 return cls.payload.is_(None)