Coverage for fss\common\service\impl\service_impl.py: 53%
47 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"""Common service impl"""
3from typing import Any, TypeVar, List, Generic, Type
5from fss.common.persistence.base_mapper import BaseMapper
6from fss.common.service.service import Service
8T = TypeVar("T", bound=Any)
9M = TypeVar("M", bound=BaseMapper)
12class ServiceImpl(Generic[M, T], Service[T]):
13 def __init__(self, mapper: Type[M]):
14 self.mapper = mapper
16 async def save(self, *, data: T) -> bool:
17 result = await self.mapper.insert(data=data)
18 return result
20 async def save_or_update(self, *, data: T) -> bool:
21 if data.id is None:
22 result = await self.mapper.insert(data=data)
23 else:
24 result = await self.update_by_id(data=data)
25 return result
27 async def save_batch(self, *, data_list: List[T]) -> bool:
28 result = await self.mapper.insert_batch(data_list=data_list)
29 return result == len(data_list)
31 async def save_or_update_batch(self, *, data_list: List[T]) -> bool:
32 if len(data_list) == 0:
33 return False
34 if data_list[0].id is None:
35 return await self.save_batch(data_list=data_list)
36 else:
37 return await self.update_batch_by_ids(data_list=data_list)
39 async def get_by_id(self, *, id: T) -> T:
40 return await self.mapper.select_by_id(id=id)
42 async def get_by_ids(self, *, ids: List[T], batch_size: int = 1000) -> List[Any]:
43 return await self.mapper.select_by_ids(ids=ids, batch_size=batch_size)
45 async def count(self) -> int:
46 return await self.mapper.select_count()
48 async def list(self, *, page: int, size: int, query: T) -> List[T]:
49 return self.mapper.select_list(page=page, size=size, query=query)
51 async def list_ordered(
52 self, *, page: int, size: int, query: T, order_by: T, sort_order: T
53 ) -> List[T]:
54 return self.mapper.select_list_ordered(
55 page=page, size=size, query=query, order_by=order_by, sort_order=sort_order
56 )
58 async def list_page(self, *, params: T, query: T) -> List[T]:
59 return self.mapper.select_list_page(params=params, query=query)
61 async def list_page_ordered(self, *, params: T, query: T, sort_order: T) -> List[T]:
62 return self.mapper.select_list_page_ordered(
63 params=params, query=query, sort_order=sort_order
64 )
66 async def update_by_id(self, *, data: T) -> bool:
67 return self.mapper.update_by_id(data=data)
69 async def update_batch_by_ids(self, *, data_list: List[T]) -> bool:
70 return self.mapper.update_batch_by_ids(data_list=data_list)
72 async def remove_by_id(self, *, id: T) -> bool:
73 return self.mapper.delete_by_id(id=id)
75 async def remove_batch_by_ids(self, *, ids: List[Any]) -> bool:
76 return self.mapper.delete_batch_by_ids(ids=ids)