Coverage for polar/meter/repository.py: 41%

1from uuid import UUID 1 ctx1a

2 

3from sqlalchemy import Select, select 1 ctx1a

4 

5from polar.auth.models import AuthSubject, Organization, User, is_organization, is_user 1 ctx1a

6from polar.kit.repository import RepositoryBase, RepositoryIDMixin 1 ctx1a

7from polar.models import Meter, UserOrganization 1 ctx1a

8 

9 

10class MeterRepository(RepositoryBase[Meter], RepositoryIDMixin[Meter, UUID]): 1 ctx1a

11 model = Meter 1 ctx1a

12 

13 async def get_readable_by_id( 1 ctx1a

14 self, id: UUID, auth_subject: AuthSubject[User | Organization] 

15 ) -> Meter | None: 

16 statement = self.get_readable_statement(auth_subject).where(Meter.id == id) 

17 return await self.get_one_or_none(statement) 

18 

19 def get_readable_statement( 1 ctx1a

20 self, auth_subject: AuthSubject[User | Organization] 

21 ) -> Select[tuple[Meter]]: 

22 statement = self.get_base_statement() 

23 

24 if is_user(auth_subject): 

25 user = auth_subject.subject 

26 statement = statement.where( 

27 Meter.organization_id.in_( 

28 select(UserOrganization.organization_id).where( 

29 UserOrganization.user_id == user.id, 

30 UserOrganization.deleted_at.is_(None), 

31 ) 

32 ) 

33 ) 

34 elif is_organization(auth_subject): 

35 statement = statement.where( 

36 Meter.organization_id == auth_subject.subject.id 

37 ) 

38 

39 return statement