Coverage for polar/models/customer_meter.py: 85%

31 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-07 12:42 +0000

1from decimal import Decimal 

2from typing import TYPE_CHECKING 

3from uuid import UUID 

4 

5from sqlalchemy import BigInteger, ForeignKey, Numeric, UniqueConstraint, Uuid 

6from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 

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

8 

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

10 

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

12 from .customer import Customer 

13 from .event import Event 

14 from .meter import Meter 

15 from .organization import Organization 

16 

17 

18class CustomerMeter(RecordModel): 

19 __tablename__ = "customer_meters" 

20 __table_args__ = (UniqueConstraint("customer_id", "meter_id"),) 

21 

22 customer_id: Mapped[UUID] = mapped_column( 

23 Uuid, ForeignKey("customers.id", ondelete="cascade") 

24 ) 

25 meter_id: Mapped[UUID] = mapped_column( 

26 Uuid, ForeignKey("meters.id", ondelete="cascade"), index=True 

27 ) 

28 last_balanced_event_id: Mapped[UUID | None] = mapped_column( 

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

30 ) 

31 consumed_units: Mapped[Decimal] = mapped_column( 

32 Numeric, nullable=False, default=0, index=True 

33 ) 

34 credited_units: Mapped[int] = mapped_column( 

35 BigInteger, nullable=False, default=0, index=True 

36 ) 

37 balance: Mapped[Decimal] = mapped_column( 

38 Numeric, nullable=False, default=Decimal(0), index=True 

39 ) 

40 

41 @declared_attr 

42 def customer(cls) -> Mapped["Customer"]: 

43 return relationship("Customer", lazy="raise_on_sql") 

44 

45 @declared_attr 

46 def meter(cls) -> Mapped["Meter"]: 

47 return relationship("Meter", lazy="raise_on_sql") 

48 

49 @declared_attr 

50 def last_balanced_event(cls) -> Mapped["Event | None"]: 

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

52 

53 organization: AssociationProxy["Organization"] = association_proxy( 

54 "customer", "organization" 

55 )