Coverage for fss\common\persistence\base_mapper.py: 67%

51 statements  

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

1"""BaseMapper defines the implemented functionalities""" 

2 

3from abc import abstractmethod 

4from typing import Any, List, Type 

5 

6from fss.common.persistence.mapper import Mapper 

7 

8 

9class BaseMapper(Mapper): 

10 @staticmethod 

11 def count_affected_rows(new_record: Any) -> int: 

12 if new_record.id is not None: 

13 return 1 

14 return 0 

15 

16 @abstractmethod 

17 def get_db_session(self) -> Type[Any]: 

18 raise NotImplementedError 

19 

20 @abstractmethod 

21 async def insert(self, *, data: Any, db_session: Any) -> int: 

22 raise NotImplementedError 

23 

24 @abstractmethod 

25 async def insert_batch(self, *, data_list: List[Any], db_session: Any) -> int: 

26 raise NotImplementedError 

27 

28 @abstractmethod 

29 async def select_by_id(self, *, id: Any, db_session: Any) -> Any: 

30 raise NotImplementedError 

31 

32 @abstractmethod 

33 async def select_by_ids( 

34 self, *, ids: List[Any], batch_size: int, db_session: Any 

35 ) -> List[Any]: 

36 raise NotImplementedError 

37 

38 @abstractmethod 

39 async def select_count(self, *, db_session: Any) -> int: 

40 raise NotImplementedError 

41 

42 @abstractmethod 

43 async def select_list( 

44 self, *, page: int, size: int, query: Any, db_session: Any 

45 ) -> List[Any]: 

46 raise NotImplementedError 

47 

48 @abstractmethod 

49 async def select_list_ordered( 

50 self, 

51 *, 

52 page: int, 

53 size: int, 

54 query: Any, 

55 order_by: Any, 

56 sort_order: Any, 

57 db_session: Any, 

58 ) -> List[Any]: 

59 raise NotImplementedError 

60 

61 @abstractmethod 

62 async def select_list_page( 

63 self, *, params: Any, query: Any, db_session: Any 

64 ) -> List[Any]: 

65 raise NotImplementedError 

66 

67 @abstractmethod 

68 async def select_list_page_ordered( 

69 self, 

70 *, 

71 params: Any, 

72 query: Any, 

73 order_by: Any, 

74 sort_order: Any, 

75 db_session: Any, 

76 ) -> List[Any]: 

77 raise NotImplementedError 

78 

79 @abstractmethod 

80 async def update_by_id(self, *, data: Any, db_session: Any) -> int: 

81 raise NotImplementedError 

82 

83 @abstractmethod 

84 async def update_batch_by_ids( 

85 self, *, data_list: List[Any], db_session: Any 

86 ) -> int: 

87 raise NotImplementedError 

88 

89 @abstractmethod 

90 async def delete_by_id(self, *, id: Any, db_session: Any) -> int: 

91 raise NotImplementedError 

92 

93 @abstractmethod 

94 async def delete_batch_by_ids(self, *, ids: List[Any], db_session: Any) -> int: 

95 raise NotImplementedError