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
« 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
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
9from polar.kit.db.models.base import RecordModel
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
18class CustomerMeter(RecordModel):
19 __tablename__ = "customer_meters"
20 __table_args__ = (UniqueConstraint("customer_id", "meter_id"),)
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 )
41 @declared_attr
42 def customer(cls) -> Mapped["Customer"]:
43 return relationship("Customer", lazy="raise_on_sql")
45 @declared_attr
46 def meter(cls) -> Mapped["Meter"]:
47 return relationship("Meter", lazy="raise_on_sql")
49 @declared_attr
50 def last_balanced_event(cls) -> Mapped["Event | None"]:
51 return relationship("Event", lazy="raise_on_sql")
53 organization: AssociationProxy["Organization"] = association_proxy(
54 "customer", "organization"
55 )