Coverage for src/lexigram/web/routing/controller.py: 46%

1"""Web controllers and routing utilities.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from typing import Any, Generic, TypeVar 

7 

8from lexigram.contracts.exceptions.domain import DomainError 

9from lexigram.contracts.web.protocols import CRUDServiceProtocol 

10from lexigram.logging import get_logger 

11from lexigram.result import Ok, Result 

12from lexigram.web.exceptions import ( 

13 NotFoundError, 

14) 

15from lexigram.web.routing.controllers import Controller 

16from lexigram.web.routing.decorators import ( 

17 delete, 

18 get, 

19 head, 

20 options, 

21 patch, 

22 post, 

23 put, 

24) 

25from lexigram.web.transport.responses import json_response 

26 

27T = TypeVar("T") 

28R = TypeVar("R", bound=Callable[..., Any]) 

29 

30 

31class GenericController(Controller, Generic[T]): 

32 """Generic controller with common CRUD patterns for any service. 

33 

34 Returns ``Result[T, DomainError]`` from all action methods. The 

35 ``ResponseSerializer`` in the request pipeline automatically maps: 

36 

37 - ``Ok(value)`` → 200 JSON response (201 for ``create_item``). 

38 - ``Err(error)`` → HTTP error via ``ResultResponseMapper``. 

39 

40 Use ``@error_status`` on your domain error classes to control the 

41 HTTP status code they map to:: 

42 

43 from lexigram.web import error_status 

44 

45 @error_status(404) 

46 class ItemNotFound(DomainError): ... 

47 """ 

48 

49 def __init__( 

50 self, 

51 service: CRUDServiceProtocol[T], 

52 resource_name: str | None = None, 

53 ) -> None: 

54 super().__init__() 

55 self.service = service 

56 self._resource_name = resource_name or self._get_default_resource_name() 

57 self._logger = get_logger(self.__class__.__name__) 

58 

59 def _get_default_resource_name(self) -> str: 

60 """Get default resource name from class name.""" 

61 return ( 

62 self.__class__.__name__.lower().replace("controller", "").replace("v1", "") 

63 ) 

64 

65 @property 

66 def resource_name(self) -> str: 

67 """Get the resource name for permissions and error messages.""" 

68 return self._resource_name 

69 

70 @property 

71 def logger(self) -> Any: 

72 """Get the logger for this controller.""" 

73 return self._logger 

74 

75 @get("/") 

76 async def list_items( 

77 self, 

78 limit: int = 20, 

79 offset: int = 0, 

80 **filters: Any, 

81 ) -> Result[dict[str, Any], DomainError]: 

82 """List items with pagination. 

83 

84 Returns a ``Result`` whose ``Ok`` payload is a dict with keys 

85 ``items``, ``limit``, ``offset``, and ``total``. 

86 """ 

87 result = await self.service.list_items(limit=limit, offset=offset, **filters) 

88 if not result.is_ok(): 

89 return result # type: ignore[return-value] 

90 items = result.unwrap() 

91 return Ok( 

92 { 

93 "items": items, 

94 "limit": limit, 

95 "offset": offset, 

96 "total": len(items), 

97 } 

98 ) 

99 

100 @get("/{item_id}") 

101 async def get_item(self, item_id: str) -> Result[T, DomainError]: 

102 """Get single item by ID.""" 

103 result = await self.service.get(item_id) 

104 if not result.is_ok(): 

105 return result # type: ignore[return-value] 

106 item = result.unwrap() 

107 if not item: 

108 raise NotFoundError(f"{self.resource_name.title()} not found") 

109 return Ok(item) 

110 

111 @post("/") 

112 async def create_item(self, data: dict[str, Any]) -> Any: 

113 """Create new item (returns 201 on success).""" 

114 result = await self.service.create(data) 

115 if not result.is_ok(): 

116 return result 

117 return json_response(result.unwrap(), status_code=201) 

118 

119 @put("/{item_id}") 

120 async def update_item( 

121 self, item_id: str, data: dict[str, Any] 

122 ) -> Result[T, DomainError]: 

123 """Update existing item.""" 

124 result = await self.service.update(item_id, data) 

125 if not result.is_ok(): 

126 return result # type: ignore[return-value] 

127 item = result.unwrap() 

128 if not item: 

129 raise NotFoundError(f"{self.resource_name.title()} not found") 

130 return Ok(item) 

131 

132 @delete("/{item_id}") 

133 async def delete_item(self, item_id: str) -> Any: 

134 """Delete item (returns 204 on success).""" 

135 result = await self.service.delete(item_id) 

136 if not result.is_ok(): 

137 return result 

138 if not result.unwrap(): 

139 raise NotFoundError(f"{self.resource_name.title()} not found") 

140 return json_response({}, status_code=204) 

141 

142 

143__all__ = [ 

144 "CRUDServiceProtocol", 

145 "Controller", 

146 "GenericController", 

147 "T", 

148 "delete", 

149 "get", 

150 "head", 

151 "options", 

152 "patch", 

153 "post", 

154 "put", 

155]