Coverage for polar/account/repository.py: 62%

1import uuid 

2from uuid import UUID 

3 

4from sqlalchemy import Select, false 

5 

6from polar.auth.models import AuthSubject, User, is_organization, is_user 

7from polar.kit.repository import ( 

8 Options, 

9 RepositoryBase, 

10 RepositorySoftDeletionIDMixin, 

11 RepositorySoftDeletionMixin, 

12) 

13from polar.models import Account, Organization 

14 

15 

16class AccountRepository( 

17 RepositorySoftDeletionIDMixin[Account, UUID], 

18 RepositorySoftDeletionMixin[Account], 

19 RepositoryBase[Account], 

20): 

21 model = Account 

22 

23 async def get_by_stripe_id( 

24 self, 

25 stripe_id: str, 

26 *, 

27 options: Options = (), 

28 include_deleted: bool = False, 

29 ) -> Account | None: 

30 statement = ( 

31 self.get_base_statement(include_deleted=include_deleted) 

32 .where(Account.stripe_id == stripe_id) 

33 .options(*options) 

34 ) 

35 return await self.get_one_or_none(statement) 

36 

37 async def get_by_user( 

38 self, user: uuid.UUID, *, options: Options = (), include_deleted: bool = False 

39 ) -> Account | None: 

40 statement = ( 

41 self.get_base_statement(include_deleted=include_deleted) 

42 .join(User, onclause=User.account_id == Account.id) 

43 .where(User.id == user) 

44 .options(*options) 

45 ) 

46 return await self.get_one_or_none(statement) 

47 

48 async def get_by_organization( 

49 self, 

50 organization: uuid.UUID, 

51 *, 

52 options: Options = (), 

53 include_deleted: bool = False, 

54 ) -> Account | None: 

55 statement = ( 

56 self.get_base_statement(include_deleted=include_deleted) 

57 .join(Organization, onclause=Organization.account_id == Account.id) 

58 .where(Organization.id == organization) 

59 .options(*options) 

60 ) 

61 return await self.get_one_or_none(statement) 

62 

63 def get_readable_statement( 

64 self, auth_subject: AuthSubject[User | Organization] 

65 ) -> Select[tuple[Account]]: 

66 statement = self.get_base_statement() 

67 

68 if is_user(auth_subject): 68 ↛ 71line 68 didn't jump to line 71 because the condition on line 68 was always true

69 user = auth_subject.subject 

70 statement = statement.where(Account.admin_id == user.id) 

71 elif is_organization(auth_subject): 

72 # Only the admin of the account can access it 

73 statement = statement.where(false()) 

74 

75 return statement