Coverage for polar/user/schemas.py: 85%

70 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 15:52 +0000

1import hashlib 1a

2import hmac 1a

3import uuid 1a

4from enum import StrEnum 1a

5from typing import Annotated, Literal 1a

6 

7from fastapi import Depends 1a

8from pydantic import UUID4, EmailStr, Field, computed_field 1a

9 

10from polar.auth.scope import Scope 1a

11from polar.config import settings 1a

12from polar.kit.schemas import Schema, TimestampedSchema, UUID4ToStr 1a

13from polar.models.user import IdentityVerificationStatus, OAuthPlatform 1a

14 

15 

16class UserBase(Schema): 1a

17 email: EmailStr 1a

18 avatar_url: str | None 1a

19 account_id: UUID4 | None 1a

20 

21 

22class OAuthAccountRead(TimestampedSchema): 1a

23 platform: OAuthPlatform 1a

24 account_id: str 1a

25 account_email: str 1a

26 account_username: str | None 1a

27 

28 

29class UserRead(UserBase, TimestampedSchema): 1a

30 id: uuid.UUID 1a

31 accepted_terms_of_service: bool 1a

32 is_admin: bool 1a

33 identity_verified: bool 1a

34 identity_verification_status: IdentityVerificationStatus 1a

35 oauth_accounts: list[OAuthAccountRead] 1a

36 

37 @computed_field 1a

38 def email_hash(self) -> str | None: 1a

39 if settings.PLAIN_CHAT_SECRET is None: 

40 return None 

41 message = hmac.new( 

42 settings.PLAIN_CHAT_SECRET.encode("utf-8"), 

43 self.email.encode("utf-8"), 

44 hashlib.sha256, 

45 ) 

46 return message.hexdigest() 

47 

48 

49class UserIdentityVerification(Schema): 1a

50 id: str 1a

51 client_secret: str 1a

52 

53 

54class UserSetAccount(Schema): 1a

55 account_id: UUID4 1a

56 

57 

58class UserStripePortalSession(Schema): 1a

59 url: str 1a

60 

61 

62class UserScopes(Schema): 1a

63 scopes: list[Scope] 1a

64 

65 

66############################################################################### 

67# USER ATTRIBUTION 

68############################################################################### 

69 

70 

71class UserSignupAttribution(Schema): 1a

72 intent: ( 1a

73 Literal[ 

74 "creator", 

75 "pledge", 

76 "purchase", 

77 "subscription", 

78 "newsletter_subscription", 

79 ] 

80 | None 

81 ) = None 

82 

83 # Flywheel sources 

84 order: UUID4ToStr | None = None 1a

85 subscription: UUID4ToStr | None = None 1a

86 pledge: UUID4ToStr | None = None 1a

87 from_storefront: UUID4ToStr | None = None 1a

88 

89 # Website source 

90 path: str | None = None 1a

91 host: str | None = None 1a

92 

93 # UTM parameters 

94 utm_source: str | None = None 1a

95 utm_medium: str | None = None 1a

96 utm_campaign: str | None = None 1a

97 

98 campaign: str | None = None 1a

99 

100 

101UserSignupAttributionQueryJSON = str | None 1a

102 

103 

104async def get_signup_attribution( 1a

105 attribution: UserSignupAttributionQueryJSON = None, 

106) -> UserSignupAttribution | None: 

107 if attribution: 

108 return UserSignupAttribution.model_validate_json(attribution) 

109 return None 

110 

111 

112UserSignupAttributionQuery = Annotated[ 1a

113 UserSignupAttribution | None, Depends(get_signup_attribution) 

114] 

115 

116 

117class UserDeletionBlockedReason(StrEnum): 1a

118 """Reasons why a user account cannot be immediately deleted.""" 

119 

120 HAS_ACTIVE_ORGANIZATIONS = "has_active_organizations" 1a

121 

122 

123class BlockingOrganization(Schema): 1a

124 """Organization that is blocking user deletion.""" 

125 

126 id: UUID4 1a

127 slug: str 1a

128 name: str 1a

129 

130 

131class UserDeletionResponse(Schema): 1a

132 """Response for user deletion request.""" 

133 

134 deleted: bool = Field( 1a

135 description="Whether the user account was immediately deleted" 

136 ) 

137 blocked_reasons: list[UserDeletionBlockedReason] = Field( 1a

138 default_factory=list, 

139 description="Reasons why immediate deletion is blocked", 

140 ) 

141 blocking_organizations: list[BlockingOrganization] = Field( 1a

142 default_factory=list, 

143 description="Organizations that must be deleted first", 

144 )