Coverage for polar/models/external_event.py: 91%

1from datetime import datetime 2 ctx1ab

2from enum import StrEnum 2 ctx1ab

3from typing import Any, Literal 2 ctx1ab

4 

5import stripe as stripe_lib 2 ctx1ab

6from sqlalchemy import ( 2 ctx1ab

7 TIMESTAMP, 

8 Boolean, 

9 ColumnElement, 

10 String, 

11 UniqueConstraint, 

12 type_coerce, 

13) 

14from sqlalchemy.dialects.postgresql import JSONB 2 ctx1ab

15from sqlalchemy.ext.hybrid import hybrid_property 2 ctx1ab

16from sqlalchemy.orm import Mapped, mapped_column 2 ctx1ab

17 

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

19from polar.kit.extensions.sqlalchemy.types import StrEnumType 2 ctx1ab

20 

21 

22class ExternalEventSource(StrEnum): 2 ctx1ab

23 stripe = "stripe" 2 ctx1ab

24 

25 

26class ExternalEvent(RecordModel): 2 ctx1ab

27 __tablename__ = "external_events" 2 ctx1ab

28 __table_args__ = (UniqueConstraint("source", "external_id"),) 2 ctx1ab

29 

30 source: Mapped[ExternalEventSource] = mapped_column( 2 ctx1ab

31 StrEnumType(ExternalEventSource), nullable=False, index=True 

32 ) 

33 handled_at: Mapped[datetime | None] = mapped_column( 2 ctx1ab

34 TIMESTAMP(timezone=True), nullable=True, default=None, index=True 

35 ) 

36 task_name: Mapped[str] = mapped_column(String, nullable=False, index=True) 2 ctx1ab

37 external_id: Mapped[str] = mapped_column(String, nullable=False, index=True) 2 ctx1ab

38 data: Mapped[dict[str, Any]] = mapped_column("data", JSONB, nullable=False) 2 ctx1ab

39 

40 @hybrid_property 2 ctx1ab

41 def is_handled(self) -> bool: 2 ctx1ab

42 return self.handled_at is not None 

43 

44 @is_handled.inplace.expression 2 ctx1ab

45 @classmethod 2 ctx1ab

46 def _is_handled_expression(cls) -> ColumnElement[bool]: 2 ctx1ab

47 return type_coerce(cls.handled_at.is_not(None), Boolean) 

48 

49 __mapper_args__ = { 2 ctx1ab

50 "polymorphic_on": "source", 

51 } 

52 

53 

54class StripeEvent(ExternalEvent): 2 ctx1ab

55 source: Mapped[Literal[ExternalEventSource.stripe]] = mapped_column( # pyright: ignore 2 ctx1ab

56 use_existing_column=True, default=ExternalEventSource.stripe 

57 ) 

58 

59 @property 2 ctx1ab

60 def stripe_data(self) -> stripe_lib.Event: 2 ctx1ab

61 return stripe_lib.Event.construct_from(self.data, key=None) 

62 

63 __mapper_args__ = { 2 ctx1ab

64 "polymorphic_identity": ExternalEventSource.stripe, 

65 "polymorphic_load": "inline", 

66 }