Coverage for polar/member/service.py: 60%

1from collections.abc import Sequence 

2from typing import Any 

3from uuid import UUID 

4 

5import structlog 

6from sqlalchemy import UnaryExpression, asc, desc 

7from sqlalchemy.exc import IntegrityError 

8 

9from polar.auth.models import AuthSubject, Organization, User 

10from polar.kit.pagination import PaginationParams 

11from polar.kit.sorting import Sorting 

12from polar.models.customer import Customer 

13from polar.models.member import Member, MemberRole 

14from polar.models.organization import Organization as OrgModel 

15from polar.postgres import AsyncReadSession, AsyncSession 

16 

17from .repository import MemberRepository 

18from .sorting import MemberSortProperty 

19 

20log = structlog.get_logger() 

21 

22 

23class MemberService: 

24 async def list( 

25 self, 

26 session: AsyncReadSession, 

27 auth_subject: AuthSubject[User | Organization], 

28 *, 

29 customer_id: UUID | None = None, 

30 pagination: PaginationParams, 

31 sorting: list[Sorting[MemberSortProperty]] = [ 

32 (MemberSortProperty.created_at, True) 

33 ], 

34 ) -> tuple[Sequence[Member], int]: 

35 """List members with pagination and filtering.""" 

36 repository = MemberRepository.from_session(session) 

37 statement = repository.get_readable_statement(auth_subject) 

38 

39 if customer_id is not None: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 statement = statement.where(Member.customer_id == customer_id) 

41 

42 order_by_clauses: list[UnaryExpression[Any]] = [] 

43 for criterion, is_desc in sorting: 

44 clause_function = desc if is_desc else asc 

45 if criterion == MemberSortProperty.created_at: 45 ↛ 43line 45 didn't jump to line 43 because the condition on line 45 was always true

46 order_by_clauses.append(clause_function(Member.created_at)) 

47 statement = statement.order_by(*order_by_clauses) 

48 

49 return await repository.paginate( 

50 statement, limit=pagination.limit, page=pagination.page 

51 ) 

52 

53 async def create_owner_member( 

54 self, 

55 session: AsyncSession, 

56 customer: Customer, 

57 organization: OrgModel, 

58 *, 

59 owner_email: str | None = None, 

60 owner_name: str | None = None, 

61 owner_external_id: str | None = None, 

62 ) -> Member | None: 

63 """ 

64 Create an owner member for a customer if feature flag is enabled. 

65 

66 Args: 

67 session: Database session 

68 customer: Customer to create member for 

69 organization: Organization the customer belongs to 

70 owner_email: Optional override for member email (defaults to customer.email) 

71 owner_name: Optional override for member name (defaults to customer.name) 

72 owner_external_id: Optional override for member external_id (defaults to customer.external_id) 

73 

74 Returns: 

75 Created/existing Member if feature flag enabled, None if flag disabled 

76 """ 

77 if not organization.feature_settings.get("member_model_enabled", False): 77 ↛ 86line 77 didn't jump to line 86 because the condition on line 77 was always true

78 log.debug( 

79 "member.create_owner_member.skipped", 

80 reason="feature_flag_disabled", 

81 customer_id=customer.id, 

82 organization_id=organization.id, 

83 ) 

84 return None 

85 

86 repository = MemberRepository.from_session(session) 

87 

88 email = owner_email or customer.email 

89 name = owner_name or customer.name 

90 external_id = owner_external_id or customer.external_id 

91 

92 existing_member = await repository.get_by_customer_and_email( 

93 session, customer, email=email 

94 ) 

95 if existing_member: 

96 log.debug( 

97 "member.create_owner_member.skipped", 

98 reason="member_already_exists", 

99 customer_id=customer.id, 

100 member_id=existing_member.id, 

101 ) 

102 return existing_member 

103 

104 member = Member( 

105 customer_id=customer.id, 

106 organization_id=organization.id, 

107 email=email, 

108 name=name, 

109 external_id=external_id, 

110 role=MemberRole.owner, 

111 ) 

112 

113 try: 

114 created_member = await repository.create(member) 

115 log.info( 

116 "member.create_owner_member.success", 

117 customer_id=customer.id, 

118 member_id=created_member.id, 

119 organization_id=organization.id, 

120 ) 

121 return created_member 

122 except IntegrityError as e: 

123 log.info( 

124 "member.create_owner_member.constraint_violation", 

125 customer_id=customer.id, 

126 organization_id=organization.id, 

127 error=str(e), 

128 reason="Likely race condition - member already exists", 

129 ) 

130 existing_member = await repository.get_by_customer_and_email( 

131 session, customer, email=email 

132 ) 

133 if existing_member: 

134 log.info( 

135 "member.create_owner_member.found_existing", 

136 customer_id=customer.id, 

137 member_id=existing_member.id, 

138 ) 

139 return existing_member 

140 

141 # Weird state: IntegrityError but member doesn't exist 

142 # Re-raise to fail customer creation and maintain data consistency 

143 log.error( 

144 "member.create_owner_member.integrity_error_no_member", 

145 customer_id=customer.id, 

146 organization_id=organization.id, 

147 error=str(e), 

148 ) 

149 raise 

150 

151 async def list_by_customer( 

152 self, 

153 session: AsyncReadSession, 

154 customer_id: UUID, 

155 ) -> Sequence[Member]: 

156 repository = MemberRepository.from_session(session) 

157 return await repository.list_by_customer(session, customer_id) 

158 

159 async def list_by_customers( 

160 self, 

161 session: AsyncReadSession, 

162 customer_ids: Sequence[UUID], 

163 ) -> Sequence[Member]: 

164 """ 

165 Get all members for multiple customers (batch loading to avoid N+1 queries). 

166 """ 

167 repository = MemberRepository.from_session(session) 

168 return await repository.list_by_customers(session, customer_ids) 

169 

170 

171member_service = MemberService()