Coverage for polar/models/meter.py: 89%

1from datetime import datetime 2 ctx1ab

2from typing import TYPE_CHECKING 2 ctx1ab

3from uuid import UUID 2 ctx1ab

4 

5from sqlalchemy import TIMESTAMP, ForeignKey, String, Uuid 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.metadata import MetadataMixin 2 ctx1ab

10from polar.meter.aggregation import Aggregation, AggregationType 2 ctx1ab

11from polar.meter.filter import Filter, FilterType 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 .event import Event 

15 from .organization import Organization 

16 

17 

18class Meter(RecordModel, MetadataMixin): 2 ctx1ab

19 __tablename__ = "meters" 2 ctx1ab

20 

21 name: Mapped[str] = mapped_column(String, nullable=False) 2 ctx1ab

22 filter: Mapped[Filter] = mapped_column(FilterType, nullable=False) 2 ctx1ab

23 aggregation: Mapped[Aggregation] = mapped_column(AggregationType, nullable=False) 2 ctx1ab

24 last_billed_event_id: Mapped[UUID | None] = mapped_column( 2 ctx1ab

25 Uuid, ForeignKey("events.id"), nullable=True, index=True, default=None 

26 ) 

27 archived_at: Mapped[datetime] = mapped_column( 2 ctx1ab

28 TIMESTAMP(timezone=True), nullable=True, default=None 

29 ) 

30 

31 @declared_attr 2 ctx1ab

32 def last_billed_event(cls) -> Mapped["Event | None"]: 2 ctx1ab

33 return relationship("Event", lazy="raise_on_sql") 2 ctx1ab

34 

35 organization_id: Mapped[UUID] = mapped_column( 2 ctx1ab

36 Uuid, 

37 ForeignKey("organizations.id", ondelete="cascade"), 

38 nullable=False, 

39 index=True, 

40 ) 

41 

42 @declared_attr 2 ctx1ab

43 def organization(cls) -> Mapped["Organization"]: 2 ctx1ab

44 return relationship("Organization", lazy="raise") 2 ctx1ab