Coverage for polar/external_event/repository.py: 31%

1from collections.abc import Sequence 

2from datetime import datetime 

3from uuid import UUID 

4 

5from polar.kit.repository import ( 

6 RepositoryBase, 

7 RepositoryIDMixin, 

8 RepositorySortingMixin, 

9 SortingClause, 

10) 

11from polar.models.external_event import ExternalEvent, ExternalEventSource 

12 

13from .sorting import ExternalEventSortProperty 

14 

15 

16class ExternalEventRepository( 

17 RepositorySortingMixin[ExternalEvent, ExternalEventSortProperty], 

18 RepositoryBase[ExternalEvent], 

19 RepositoryIDMixin[ExternalEvent, UUID], 

20): 

21 model = ExternalEvent 

22 

23 async def get_by_source_and_id( 

24 self, source: ExternalEventSource, id: UUID 

25 ) -> ExternalEvent | None: 

26 statement = self.get_base_statement().where( 

27 ExternalEvent.source == source, ExternalEvent.id == id 

28 ) 

29 return await self.get_one_or_none(statement) 

30 

31 async def get_by_source_and_external_id( 

32 self, source: ExternalEventSource, external_id: str 

33 ) -> ExternalEvent | None: 

34 statement = self.get_base_statement().where( 

35 ExternalEvent.source == source, ExternalEvent.external_id == external_id 

36 ) 

37 return await self.get_one_or_none(statement) 

38 

39 async def get_all_unhandled( 

40 self, older_than: datetime | None = None 

41 ) -> Sequence[ExternalEvent]: 

42 statement = self.get_base_statement().where(ExternalEvent.handled_at.is_(None)) 

43 if older_than is not None: 

44 statement = statement.where(ExternalEvent.created_at < older_than) 

45 return await self.get_all(statement) 

46 

47 def get_sorting_clause(self, property: ExternalEventSortProperty) -> SortingClause: 

48 match property: 

49 case ExternalEventSortProperty.created_at: 

50 return self.model.created_at 

51 case ExternalEventSortProperty.handled_at: 

52 return self.model.handled_at 

53 case ExternalEventSortProperty.source: 

54 return self.model.source 

55 case ExternalEventSortProperty.task_name: 

56 return self.model.task_name