Coverage for polar/account/repository.py: 41%
25 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1import uuid 1a
2from uuid import UUID 1a
4from sqlalchemy import Select, false 1a
6from polar.auth.models import AuthSubject, User, is_organization, is_user 1a
7from polar.kit.repository import ( 1a
8 Options,
9 RepositoryBase,
10 RepositorySoftDeletionIDMixin,
11 RepositorySoftDeletionMixin,
12)
13from polar.models import Account, Organization 1a
16class AccountRepository( 1a
17 RepositorySoftDeletionIDMixin[Account, UUID],
18 RepositorySoftDeletionMixin[Account],
19 RepositoryBase[Account],
20):
21 model = Account 1a
23 async def get_by_stripe_id( 1a
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)
37 async def get_by_user( 1a
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)
48 async def get_by_organization( 1a
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)
63 def get_readable_statement( 1a
64 self, auth_subject: AuthSubject[User | Organization]
65 ) -> Select[tuple[Account]]:
66 statement = self.get_base_statement()
68 if is_user(auth_subject):
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())
75 return statement