Coverage for polar/file/repository.py: 34%

31 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 17:15 +0000

1from collections.abc import Sequence 1a

2from uuid import UUID 1a

3 

4from sqlalchemy import Select, select 1a

5 

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

7from polar.kit.repository import ( 1a

8 RepositoryBase, 

9 RepositorySoftDeletionIDMixin, 

10 RepositorySoftDeletionMixin, 

11 RepositorySortingMixin, 

12 SortingClause, 

13) 

14from polar.kit.sorting import Sorting 1a

15from polar.models import File, UserOrganization 1a

16from polar.models.file import FileServiceTypes 1a

17 

18from .sorting import FileSortProperty 1a

19 

20 

21class FileRepository( 1a

22 RepositorySortingMixin[File, FileSortProperty], 

23 RepositorySoftDeletionIDMixin[File, UUID], 

24 RepositorySoftDeletionMixin[File], 

25 RepositoryBase[File], 

26): 

27 model = File 1a

28 

29 def get_readable_statement( 1a

30 self, auth_subject: AuthSubject[User | Organization] 

31 ) -> Select[tuple[File]]: 

32 statement = self.get_base_statement() 

33 

34 if is_user(auth_subject): 

35 user = auth_subject.subject 

36 statement = statement.where( 

37 File.organization_id.in_( 

38 select(UserOrganization.organization_id).where( 

39 UserOrganization.user_id == user.id, 

40 UserOrganization.deleted_at.is_(None), 

41 ) 

42 ) 

43 ) 

44 elif is_organization(auth_subject): 

45 statement = statement.where( 

46 File.organization_id == auth_subject.subject.id, 

47 ) 

48 

49 return statement 

50 

51 async def get_all_by_organization( 1a

52 self, 

53 organization_id: UUID, 

54 *, 

55 service: FileServiceTypes | None = None, 

56 sorting: list[Sorting[FileSortProperty]] = [ 

57 (FileSortProperty.created_at, True) 

58 ], 

59 ) -> Sequence[File]: 

60 """Get all files for an organization, optionally filtered by service type.""" 

61 statement = self.get_base_statement().where( 

62 File.organization_id == organization_id, 

63 File.is_uploaded.is_(True), 

64 ) 

65 

66 if service is not None: 

67 statement = statement.where(File.service == service) 

68 

69 statement = self.apply_sorting(statement, sorting) 

70 

71 return await self.get_all(statement) 

72 

73 def get_sorting_clause(self, property: FileSortProperty) -> SortingClause: 1a

74 match property: 

75 case FileSortProperty.created_at: 

76 return File.created_at 

77 case FileSortProperty.file_name: 

78 return File.name