Coverage for app/backend/src/couchers/models/donations.py: 100%

1import enum 

2from datetime import datetime 

3from typing import TYPE_CHECKING 

4 

5from sqlalchemy import BigInteger, ColumnElement, DateTime, Enum, Float, ForeignKey, Integer, String, func 

6from sqlalchemy.orm import Mapped, mapped_column, relationship 

7 

8from couchers.models.base import Base 

9 

10if TYPE_CHECKING: 

11 from couchers.models.users import User 

12 

13 

14class DonationType(enum.Enum): 

15 one_time = enum.auto() 

16 monthly = enum.auto() 

17 yearly = enum.auto() 

18 

19 

20class InvoiceType(enum.Enum): 

21 on_platform = enum.auto() 

22 external_shop = enum.auto() 

23 

24 

25class DonationInitiation(Base, kw_only=True): 

26 """ 

27 Whenever someone initiates a donation through the platform 

28 """ 

29 

30 __tablename__ = "donation_initiations" 

31 

32 # reserved up front with next_id() 

33 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=True) 

34 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

35 user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True) 

36 

37 amount: Mapped[int] = mapped_column(Integer) 

38 stripe_checkout_session_id: Mapped[str] = mapped_column(String) 

39 

40 donation_type: Mapped[DonationType] = mapped_column(Enum(DonationType)) 

41 source: Mapped[str | None] = mapped_column(String, default=None) 

42 

43 user: Mapped[User] = relationship(init=False, backref="donation_initiations") 

44 

45 @classmethod 

46 def next_id(cls) -> ColumnElement[int]: 

47 """ 

48 Reserve the next value from this table's id sequence, you can call this at the start of the transaction, then 

49 do something with it, and insert it later as the primary key 

50 """ 

51 return func.nextval(func.pg_get_serial_sequence(cls.__tablename__, "id")) 

52 

53 

54class Invoice(Base, kw_only=True): 

55 """ 

56 Successful donations, both one-off and recurring 

57 

58 Triggered by `payment_intent.succeeded` webhook 

59 """ 

60 

61 __tablename__ = "invoices" 

62 

63 id: Mapped[int] = mapped_column(BigInteger, primary_key=True, init=False) 

64 created: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), init=False) 

65 user_id: Mapped[int] = mapped_column(ForeignKey("users.id")) 

66 

67 amount: Mapped[float] = mapped_column(Float) 

68 

69 stripe_payment_intent_id: Mapped[str] = mapped_column(String, unique=True) 

70 stripe_receipt_url: Mapped[str] = mapped_column(String) 

71 

72 invoice_type: Mapped[InvoiceType] = mapped_column(Enum(InvoiceType)) 

73 

74 # null for payments that predate us passing the initiation id to Stripe 

75 donation_initiation_id: Mapped[int | None] = mapped_column( 

76 ForeignKey("donation_initiations.id"), index=True, default=None 

77 ) 

78 donation_type: Mapped[DonationType | None] = mapped_column(Enum(DonationType), default=None) 

79 

80 user: Mapped[User] = relationship(init=False, backref="invoices")