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

1from decimal import Decimal 2 ctx1ab

2from typing import TYPE_CHECKING 2 ctx1ab

3from uuid import UUID 2 ctx1ab

4 

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

6from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 2 ctx1ab

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

8 

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

10 

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

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): 2 ctx1ab

19 __tablename__ = "customer_meters" 2 ctx1ab

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

21 

22 customer_id: Mapped[UUID] = mapped_column( 2 ctx1ab

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

24 ) 

25 meter_id: Mapped[UUID] = mapped_column( 2 ctx1ab

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

27 ) 

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

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

30 ) 

31 consumed_units: Mapped[Decimal] = mapped_column( 2 ctx1ab

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

33 ) 

34 credited_units: Mapped[int] = mapped_column( 2 ctx1ab

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

36 ) 

37 balance: Mapped[Decimal] = mapped_column( 2 ctx1ab

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

39 ) 

40 

41 @declared_attr 2 ctx1ab

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

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

44 

45 @declared_attr 2 ctx1ab

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

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

48 

49 @declared_attr 2 ctx1ab

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

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

52 

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

54 "customer", "organization" 

55 )