Coverage for src/auth/api/role_controller.py: 100%

1from typing import Annotated 

2 

3from advanced_alchemy.extensions.litestar.dto import SQLAlchemyDTO, SQLAlchemyDTOConfig 

4from advanced_alchemy.extensions.litestar.providers import create_service_dependencies 

5from advanced_alchemy.filters import FilterTypes 

6from advanced_alchemy.service import OffsetPagination 

7from litestar import delete, get, post 

8from litestar.controller import Controller 

9from litestar.params import Dependency 

10import msgspec 

11 

12from ..models import RoleModel 

13from ..services import RoleService 

14 

15 

16class RoleCreateRequest(msgspec.Struct): 

17 name: str 

18 

19 

20class RoleDTO(SQLAlchemyDTO[RoleModel]): 

21 config = SQLAlchemyDTOConfig() 

22 

23 

24class RoleController(Controller): 

25 dependencies = create_service_dependencies( 

26 RoleService, 

27 key="service", 

28 filters={ 

29 "created_at": True, 

30 "updated_at": True, 

31 }, 

32 ) 

33 return_dto = RoleDTO 

34 

35 @get(operation_id="ListRoles", path="/roles") 

36 async def list_roles( 

37 self, 

38 service: RoleService, 

39 filters: Annotated[list[FilterTypes], Dependency(skip_validation=True)], 

40 ) -> OffsetPagination[RoleModel]: 

41 results, total = await service.list_and_count(*filters) 

42 return service.to_schema(data=results, total=total, filters=filters) 

43 

44 @post(operation_id="CreateRole", path="/roles") 

45 async def create_role(self, service: RoleService, data: RoleCreateRequest) -> RoleModel: 

46 return await service.create(RoleModel(**msgspec.to_builtins(data)), auto_commit=True) 

47 

48 @delete(operation_id="DeleteRole", path="/roles/{item_id:str}") 

49 async def delete_role(self, service: RoleService, item_id: str) -> None: 

50 await service.delete(item_id, auto_commit=True)