Coverage for fss\common\persistence\base_mapper.py: 67%
51 statements
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-11 19:09 +0800
« prev ^ index » next coverage.py v7.4.4, created at 2024-04-11 19:09 +0800
1"""BaseMapper defines the implemented functionalities"""
3from abc import abstractmethod
4from typing import Any, List, Type
6from fss.common.persistence.mapper import Mapper
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
16 @abstractmethod
17 def get_db_session(self) -> Type[Any]:
18 raise NotImplementedError
20 @abstractmethod
21 async def insert(self, *, data: Any, db_session: Any) -> int:
22 raise NotImplementedError
24 @abstractmethod
25 async def insert_batch(self, *, data_list: List[Any], db_session: Any) -> int:
26 raise NotImplementedError
28 @abstractmethod
29 async def select_by_id(self, *, id: Any, db_session: Any) -> Any:
30 raise NotImplementedError
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
38 @abstractmethod
39 async def select_count(self, *, db_session: Any) -> int:
40 raise NotImplementedError
42 @abstractmethod
43 async def select_list(
44 self, *, page: int, size: int, query: Any, db_session: Any
45 ) -> List[Any]:
46 raise NotImplementedError
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
61 @abstractmethod
62 async def select_list_page(
63 self, *, params: Any, query: Any, db_session: Any
64 ) -> List[Any]:
65 raise NotImplementedError
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
79 @abstractmethod
80 async def update_by_id(self, *, data: Any, db_session: Any) -> int:
81 raise NotImplementedError
83 @abstractmethod
84 async def update_batch_by_ids(
85 self, *, data_list: List[Any], db_session: Any
86 ) -> int:
87 raise NotImplementedError
89 @abstractmethod
90 async def delete_by_id(self, *, id: Any, db_session: Any) -> int:
91 raise NotImplementedError
93 @abstractmethod
94 async def delete_batch_by_ids(self, *, ids: List[Any], db_session: Any) -> int:
95 raise NotImplementedError