Coverage for polar/models/account.py: 68%

1from enum import StrEnum 2 ctx1ab

2from typing import TYPE_CHECKING, Any, TypeAlias 2 ctx1ab

3from uuid import UUID 2 ctx1ab

4 

5from sqlalchemy import Boolean, ForeignKey, Integer, String, Text, Uuid 2 ctx1ab

6from sqlalchemy.dialects.postgresql import JSONB 2 ctx1ab

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

8 

9from polar.config import settings 2 ctx1ab

10from polar.enums import AccountType 2 ctx1ab

11from polar.kit.address import Address, AddressType 2 ctx1ab

12from polar.kit.db.models import RecordModel 2 ctx1ab

13from polar.kit.extensions.sqlalchemy import StringEnum 2 ctx1ab

14from polar.kit.math import polar_round 2 ctx1ab

15 

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

17 from .organization import Organization 

18 from .user import User 

19 

20 

21FeeBasisPoints: TypeAlias = int 2 ctx1ab

22FeeFixedCents: TypeAlias = int 2 ctx1ab

23Fees: TypeAlias = tuple[FeeBasisPoints, FeeFixedCents] 2 ctx1ab

24 

25 

26class Account(RecordModel): 2 ctx1ab

27 class Status(StrEnum): 2 ctx1ab

28 CREATED = "created" 2 ctx1ab

29 ONBOARDING_STARTED = "onboarding_started" 2 ctx1ab

30 UNDER_REVIEW = "under_review" 2 ctx1ab

31 DENIED = "denied" 2 ctx1ab

32 ACTIVE = "active" 2 ctx1ab

33 

34 def get_display_name(self) -> str: 2 ctx1ab

35 return { 

36 Account.Status.CREATED: "Created", 

37 Account.Status.ONBOARDING_STARTED: "Onboarding Started", 

38 Account.Status.UNDER_REVIEW: "Under Review", 

39 Account.Status.DENIED: "Denied", 

40 Account.Status.ACTIVE: "Active", 

41 }[self] 

42 

43 __tablename__ = "accounts" 2 ctx1ab

44 

45 account_type: Mapped[AccountType] = mapped_column( 2 ctx1ab

46 StringEnum(AccountType), nullable=False 

47 ) 

48 

49 admin_id: Mapped[UUID] = mapped_column(Uuid, ForeignKey("users.id", use_alter=True)) 2 ctx1ab

50 

51 stripe_id: Mapped[str | None] = mapped_column( 2 ctx1ab

52 String(100), nullable=True, default=None 

53 ) 

54 open_collective_slug: Mapped[str | None] = mapped_column( 2 ctx1ab

55 String(255), nullable=True, default=None 

56 ) 

57 

58 email: Mapped[str | None] = mapped_column(String(254), nullable=True, default=None) 2 ctx1ab

59 

60 country: Mapped[str] = mapped_column(String(2), nullable=False) 2 ctx1ab

61 currency: Mapped[str] = mapped_column(String(3)) 2 ctx1ab

62 

63 is_details_submitted: Mapped[bool] = mapped_column(Boolean, nullable=False) 2 ctx1ab

64 is_charges_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False) 2 ctx1ab

65 is_payouts_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False) 2 ctx1ab

66 

67 processor_fees_applicable: Mapped[bool] = mapped_column( 2 ctx1ab

68 Boolean, nullable=False, default=True 

69 ) 

70 _platform_fee_percent: Mapped[int | None] = mapped_column( 2 ctx1ab

71 Integer, name="platform_fee_percent", nullable=True, default=None 

72 ) 

73 _platform_fee_fixed: Mapped[int | None] = mapped_column( 2 ctx1ab

74 Integer, name="platform_fee_fixed", nullable=True, default=None 

75 ) 

76 

77 business_type: Mapped[str | None] = mapped_column( 2 ctx1ab

78 String(255), nullable=True, default=None 

79 ) 

80 

81 status: Mapped[Status] = mapped_column( 2 ctx1ab

82 StringEnum(Status), nullable=False, default=Status.CREATED 

83 ) 

84 next_review_threshold: Mapped[int | None] = mapped_column( 2 ctx1ab

85 Integer, nullable=True, default=0 

86 ) 

87 campaign_id: Mapped[UUID | None] = mapped_column( 2 ctx1ab

88 Uuid, 

89 ForeignKey("campaigns.id", ondelete="set null"), 

90 default=None, 

91 index=True, 

92 ) 

93 

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

95 

96 billing_name: Mapped[str | None] = mapped_column( 2 ctx1ab

97 String, nullable=True, default=None 

98 ) 

99 billing_address: Mapped[Address | None] = mapped_column(AddressType, nullable=True) 2 ctx1ab

100 billing_additional_info: Mapped[str | None] = mapped_column( 2 ctx1ab

101 Text, nullable=True, default=None 

102 ) 

103 billing_notes: Mapped[str | None] = mapped_column(Text, nullable=True, default=None) 2 ctx1ab

104 

105 @declared_attr 2 ctx1ab

106 def admin(cls) -> Mapped["User"]: 2 ctx1ab

107 return relationship("User", lazy="raise", foreign_keys="[Account.admin_id]") 2 ctx1ab

108 

109 @declared_attr 2 ctx1ab

110 def users(cls) -> Mapped[list["User"]]: 2 ctx1ab

111 return relationship( 2 ctx1ab

112 "User", 

113 lazy="raise", 

114 back_populates="account", 

115 foreign_keys="[User.account_id]", 

116 ) 

117 

118 @declared_attr 2 ctx1ab

119 def all_organizations(cls) -> Mapped[list["Organization"]]: 2 ctx1ab

120 return relationship("Organization", lazy="raise", back_populates="account") 2 ctx1ab

121 

122 @declared_attr 2 ctx1ab

123 def organizations(cls) -> Mapped[list["Organization"]]: 2 ctx1ab

124 return relationship( 2 ctx1ab

125 "Organization", 

126 lazy="raise", 

127 primaryjoin=( 

128 "and_(" 

129 "Organization.account_id == Account.id," 

130 "Organization.deleted_at.is_(None)" 

131 ")" 

132 ), 

133 viewonly=True, 

134 ) 

135 

136 def is_active(self) -> bool: 2 ctx1ab

137 return self.status == Account.Status.ACTIVE 

138 

139 def is_under_review(self) -> bool: 2 ctx1ab

140 return self.status == Account.Status.UNDER_REVIEW 

141 

142 def is_payout_ready(self) -> bool: 2 ctx1ab

143 return self.is_active() and ( 

144 # For Stripe accounts, check if payouts are enabled. 

145 # Normally, the account shouldn't be active if payouts are not enabled 

146 # but let's be extra cautious 

147 self.account_type != AccountType.stripe or self.is_payouts_enabled 

148 ) 

149 

150 def get_associations_names(self) -> list[str]: 2 ctx1ab

151 associations_names: list[str] = [] 

152 for user in self.users: 

153 associations_names.append(user.email) 

154 for organization in self.organizations: 

155 associations_names.append(organization.slug) 

156 return associations_names 

157 

158 @property 2 ctx1ab

159 def platform_fee(self) -> Fees: 2 ctx1ab

160 basis_points = self._platform_fee_percent 

161 if basis_points is None: 

162 basis_points = settings.PLATFORM_FEE_BASIS_POINTS 

163 

164 fixed = self._platform_fee_fixed 

165 if fixed is None: 

166 fixed = settings.PLATFORM_FEE_FIXED 

167 

168 return (basis_points, fixed) 

169 

170 def calculate_fee_in_cents(self, amount_in_cents: int) -> int: 2 ctx1ab

171 basis_points, fixed = self.platform_fee 

172 fee_in_cents = (amount_in_cents * (basis_points / 10_000)) + fixed 

173 # Apply same logic as Stripe fee rounding 

174 return polar_round(fee_in_cents)