Coverage for polar/models/processor_transaction.py: 96%

1import datetime 2 ctx1ab

2from enum import StrEnum 2 ctx1ab

3from typing import Any, Self 2 ctx1ab

4from uuid import UUID 2 ctx1ab

5 

6import stripe as stripe_lib 2 ctx1ab

7from sqlalchemy import TIMESTAMP, BigInteger, String, Text, Uuid 2 ctx1ab

8from sqlalchemy.dialects.postgresql import JSONB 2 ctx1ab

9from sqlalchemy.orm import Mapped, mapped_column 2 ctx1ab

10 

11from polar.kit.db.models import Model 2 ctx1ab

12from polar.kit.extensions.sqlalchemy.types import StringEnum 2 ctx1ab

13from polar.kit.utils import generate_uuid, utc_now 2 ctx1ab

14 

15 

16class Processor(StrEnum): 2 ctx1ab

17 """ 

18 Supported payment or payout processors, i.e rails for transactions. 

19 """ 

20 

21 stripe = "stripe" 2 ctx1ab

22 

23 

24class ProcessorTransaction(Model): 2 ctx1ab

25 __tablename__ = "processor_transactions" 2 ctx1ab

26 

27 id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=generate_uuid) 2 ctx1ab

28 timestamp: Mapped[datetime.datetime] = mapped_column( 2 ctx1ab

29 TIMESTAMP(timezone=True), nullable=False, default=utc_now, index=True 

30 ) 

31 processor: Mapped[Processor] = mapped_column( 2 ctx1ab

32 StringEnum(Processor), index=True, nullable=False 

33 ) 

34 processor_id: Mapped[str] = mapped_column( 2 ctx1ab

35 String, nullable=False, index=True, unique=True 

36 ) 

37 type: Mapped[str] = mapped_column(String, nullable=False, index=True) 2 ctx1ab

38 currency: Mapped[str] = mapped_column(String(3), nullable=False) 2 ctx1ab

39 amount: Mapped[int] = mapped_column(BigInteger, nullable=False) 2 ctx1ab

40 fee: Mapped[int] = mapped_column(BigInteger, nullable=False) 2 ctx1ab

41 description: Mapped[str | None] = mapped_column(Text, nullable=True) 2 ctx1ab

42 raw: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) 2 ctx1ab

43 

44 @classmethod 2 ctx1ab

45 def from_stripe(cls, bt: stripe_lib.BalanceTransaction) -> Self: 2 ctx1ab

46 return cls( 

47 timestamp=datetime.datetime.fromtimestamp(bt.created, tz=datetime.UTC), 

48 processor=Processor.stripe, 

49 processor_id=bt.id, 

50 type=bt.type, 

51 currency=bt.currency, 

52 amount=bt.amount, 

53 fee=bt.fee, 

54 description=bt.description, 

55 raw=bt, 

56 )