Coverage for fss\common\service\service.py: 70%

50 statements  

« prev     ^ index     » next       coverage.py v7.4.4, created at 2024-04-12 22:20 +0800

1"""Abstract Service used in the project""" 

2 

3from abc import ABC, abstractmethod 

4from typing import Any, List, TypeVar, Generic 

5 

6T = TypeVar("T", bound=Any) 

7DEFAULT_BATCH_SIZE: int = 1000 

8 

9 

10class Service(Generic[T], ABC): 

11 @abstractmethod 

12 async def save(self, *, data: T) -> bool: 

13 raise NotImplementedError 

14 

15 @abstractmethod 

16 async def save_or_update(self, *, data: T) -> bool: 

17 raise NotImplementedError 

18 

19 @abstractmethod 

20 async def save_batch(self, *, data_list: List[T]) -> bool: 

21 raise NotImplementedError 

22 

23 @abstractmethod 

24 async def save_or_update_batch(self, *, data_list: List[T]) -> bool: 

25 raise NotImplementedError 

26 

27 @abstractmethod 

28 async def get_by_id(self, *, id: T) -> T: 

29 raise NotImplementedError 

30 

31 @abstractmethod 

32 async def get_by_ids(self, *, ids: List[T], batch_size: int) -> List[Any]: 

33 raise NotImplementedError 

34 

35 @abstractmethod 

36 async def count( 

37 self, 

38 ) -> int: 

39 raise NotImplementedError 

40 

41 @abstractmethod 

42 async def list(self, *, page: int, size: int, query: T) -> List[T]: 

43 raise NotImplementedError 

44 

45 @abstractmethod 

46 async def list_ordered( 

47 self, *, page: int, size: int, query: T, order_by: T, sort_order: T 

48 ) -> List[T]: 

49 raise NotImplementedError 

50 

51 @abstractmethod 

52 async def list_page(self, *, params: T, query: T) -> List[T]: 

53 raise NotImplementedError 

54 

55 @abstractmethod 

56 async def list_page_ordered(self, *, params: T, query: T, sort_order: T) -> List[T]: 

57 raise NotImplementedError 

58 

59 @abstractmethod 

60 async def update_by_id(self, *, data: T) -> bool: 

61 raise NotImplementedError 

62 

63 @abstractmethod 

64 async def update_batch_by_ids(self, *, data_list: List[T]) -> bool: 

65 raise NotImplementedError 

66 

67 @abstractmethod 

68 async def remove_by_id(self, *, id: T) -> bool: 

69 raise NotImplementedError 

70 

71 @abstractmethod 

72 async def remove_batch_by_ids(self, *, ids: List[Any]) -> bool: 

73 raise NotImplementedError