base_service.py 1.3 KB

1234567891011121314151617181920212223242526272829303132333435363738
  1. from typing import Any, List, Dict, TypeVar, Generic
  2. from BACKEND.domain.dtos.base_dto import DtoBase, CreateDtoBase, UpdateDtoBase
  3. # 定义泛型类型变量
  4. GetSchemaType = TypeVar("GetSchemaType", bound=DtoBase)
  5. CreateSchemaType = TypeVar("CreateSchemaType", bound=CreateDtoBase)
  6. UpdateSchemaType = TypeVar("UpdateSchemaType", bound=UpdateDtoBase)
  7. class ServiceBase(Generic[GetSchemaType, CreateSchemaType, UpdateSchemaType]):
  8. __abstract__ = True
  9. def __init__(self, store_class):
  10. self.store_class = store_class
  11. def get(self, id: int, include_deleted: bool = False) -> GetSchemaType:
  12. model = self.store_class.get(id, include_deleted)
  13. return model.from_model()
  14. def list(
  15. self,
  16. page: int = 1,
  17. page_size: int = 20,
  18. keyword: str = None,
  19. filters: List = None,
  20. order: str = None,
  21. ) -> Dict[str, Any]:
  22. return self.store_class.list(page, page_size, keyword, filters, order)
  23. def create(self, obj_in: CreateSchemaType) -> Any:
  24. return self.store_class.create(obj_in.model_dump())
  25. def update(self, obj_in: UpdateSchemaType) -> Any:
  26. return self.store_class.update(obj_in.model_dump())
  27. def delete(self, id: int) -> None:
  28. return self.store_class.delete(id)