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

1import hashlib 

2import hmac 

3import uuid 

4from enum import StrEnum 

5from typing import Annotated, Literal 

6 

7from fastapi import Depends 

8from pydantic import UUID4, EmailStr, Field, computed_field 

9 

10from polar.auth.scope import Scope 

11from polar.config import settings 

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

13from polar.models.user import IdentityVerificationStatus, OAuthPlatform 

14 

15 

16class UserBase(Schema): 

17 email: EmailStr 

18 avatar_url: str | None 

19 account_id: UUID4 | None 

20 

21 

22class OAuthAccountRead(TimestampedSchema): 

23 platform: OAuthPlatform 

24 account_id: str 

25 account_email: str 

26 account_username: str | None 

27 

28 

29class UserRead(UserBase, TimestampedSchema): 

30 id: uuid.UUID 

31 accepted_terms_of_service: bool 

32 is_admin: bool 

33 identity_verified: bool 

34 identity_verification_status: IdentityVerificationStatus 

35 oauth_accounts: list[OAuthAccountRead] 

36 

37 @computed_field 

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

39 if settings.PLAIN_CHAT_SECRET is None: 39 ↛ 41line 39 didn't jump to line 41 because the condition on line 39 was always true

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): 

50 id: str 

51 client_secret: str 

52 

53 

54class UserSetAccount(Schema): 

55 account_id: UUID4 

56 

57 

58class UserStripePortalSession(Schema): 

59 url: str 

60 

61 

62class UserScopes(Schema): 

63 scopes: list[Scope] 

64 

65 

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

67# USER ATTRIBUTION 

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

69 

70 

71class UserSignupAttribution(Schema): 

72 intent: ( 

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 

85 subscription: UUID4ToStr | None = None 

86 pledge: UUID4ToStr | None = None 

87 from_storefront: UUID4ToStr | None = None 

88 

89 # Website source 

90 path: str | None = None 

91 host: str | None = None 

92 

93 # UTM parameters 

94 utm_source: str | None = None 

95 utm_medium: str | None = None 

96 utm_campaign: str | None = None 

97 

98 campaign: str | None = None 

99 

100 

101UserSignupAttributionQueryJSON = str | None 

102 

103 

104async def get_signup_attribution( 

105 attribution: UserSignupAttributionQueryJSON = None, 

106) -> UserSignupAttribution | None: 

107 if attribution: 

108 return UserSignupAttribution.model_validate_json(attribution) 

109 return None 

110 

111 

112UserSignupAttributionQuery = Annotated[ 

113 UserSignupAttribution | None, Depends(get_signup_attribution) 

114] 

115 

116 

117class UserDeletionBlockedReason(StrEnum): 

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

119 

120 HAS_ACTIVE_ORGANIZATIONS = "has_active_organizations" 

121 

122 

123class BlockingOrganization(Schema): 

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

125 

126 id: UUID4 

127 slug: str 

128 name: str 

129 

130 

131class UserDeletionResponse(Schema): 

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

133 

134 deleted: bool = Field( 

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

136 ) 

137 blocked_reasons: list[UserDeletionBlockedReason] = Field( 

138 default_factory=list, 

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

140 ) 

141 blocking_organizations: list[BlockingOrganization] = Field( 

142 default_factory=list, 

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

144 )