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

1from datetime import datetime 

2from typing import TYPE_CHECKING 

3from uuid import UUID 

4 

5from sqlalchemy import TIMESTAMP, ForeignKey, String, Uuid 

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

7 

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

9from polar.kit.metadata import MetadataMixin 

10from polar.meter.aggregation import Aggregation, AggregationType 

11from polar.meter.filter import Filter, FilterType 

12 

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

14 from .event import Event 

15 from .organization import Organization 

16 

17 

18class Meter(RecordModel, MetadataMixin): 

19 __tablename__ = "meters" 

20 

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

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

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

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

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

26 ) 

27 archived_at: Mapped[datetime] = mapped_column( 

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

29 ) 

30 

31 @declared_attr 

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

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

34 

35 organization_id: Mapped[UUID] = mapped_column( 

36 Uuid, 

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

38 nullable=False, 

39 index=True, 

40 ) 

41 

42 @declared_attr 

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

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