Coverage for src/lexigram/web/pipes/builtin/parse.py: 27%

75 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-25 04:37 +0800

1"""Parse pipes for type conversion. 

2 

3Pipes that parse string values into specific types. 

4""" 

5 

6from __future__ import annotations 

7 

8from datetime import datetime 

9from typing import Any 

10import uuid 

11 

12from lexigram.web.protocols import ParamMetadata, PipeProtocol 

13 

14 

15class ParseIntPipe(PipeProtocol): 

16 """PipeProtocol that parses string values to integers. 

17 

18 Example: 

19 ```python 

20 class UserController(Controller): 

21 @get("/users/{user_id}") 

22 async def get_user(self, @path(pipe=ParseIntPipe()) user_id: int): 

23 ... 

24 ``` 

25 """ 

26 

27 def __init__(self, strict: bool = True): 

28 """Initialize the pipe. 

29 

30 Args: 

31 strict: If True, raises error on invalid input. If False, returns default. 

32 """ 

33 self._strict = strict 

34 

35 async def transform(self, value: Any, metadata: ParamMetadata) -> int: 

36 """Parse value to integer. 

37 

38 Args: 

39 value: The value to parse. 

40 metadata: Metadata about the parameter. 

41 

42 Returns: 

43 Parsed integer. 

44 

45 Raises: 

46 ValueError: If strict=True and value cannot be parsed. 

47 """ 

48 if value is None: 

49 if metadata.default is not None: 

50 return int(metadata.default) 

51 if self._strict: 

52 raise ValueError(f"Missing required parameter: {metadata.name}") 

53 return 0 

54 

55 try: 

56 return int(value) 

57 except (ValueError, TypeError) as e: 

58 if self._strict: 

59 raise ValueError(f"Invalid integer for {metadata.name}: {value}") from e 

60 return int(metadata.default) if metadata.default is not None else 0 

61 

62 

63class ParseUUIDPipe(PipeProtocol): 

64 """PipeProtocol that parses string values to UUIDs. 

65 

66 Example: 

67 ```python 

68 class UserController(Controller): 

69 @get("/users/{user_id}") 

70 async def get_user(self, @path(pipe=ParseUUIDPipe()) user_id: uuid.UUID): 

71 ... 

72 ``` 

73 """ 

74 

75 async def transform(self, value: Any, metadata: ParamMetadata) -> uuid.UUID: 

76 """Parse value to UUID. 

77 

78 Args: 

79 value: The value to parse. 

80 metadata: Metadata about the parameter. 

81 

82 Returns: 

83 Parsed UUID. 

84 

85 Raises: 

86 ValueError: If value cannot be parsed as UUID. 

87 """ 

88 if value is None: 

89 if metadata.default is not None: 

90 return uuid.UUID(str(metadata.default)) 

91 raise ValueError(f"Missing required parameter: {metadata.name}") 

92 

93 try: 

94 return uuid.UUID(str(value)) 

95 except (ValueError, TypeError) as e: 

96 raise ValueError(f"Invalid UUID for {metadata.name}: {value}") from e 

97 

98 

99class ParseBoolPipe(PipeProtocol): 

100 """PipeProtocol that parses string values to booleans. 

101 

102 Accepts: "true", "false", "1", "0", "yes", "no" 

103 

104 Example: 

105 ```python 

106 class UserController(Controller): 

107 @get("/users") 

108 async def list_users(self, @query(pipe=ParseBoolPipe()) active: bool = True): 

109 ... 

110 ``` 

111 """ 

112 

113 TRUE_VALUES = {"true", "1", "yes", "on"} 

114 FALSE_VALUES = {"false", "0", "no", "off"} 

115 

116 async def transform(self, value: Any, metadata: ParamMetadata) -> bool: 

117 """Parse value to boolean. 

118 

119 Args: 

120 value: The value to parse. 

121 metadata: Metadata about the parameter. 

122 

123 Returns: 

124 Parsed boolean. 

125 """ 

126 if value is None: 

127 return metadata.default if metadata.default is not None else True 

128 

129 if isinstance(value, bool): 

130 return value 

131 

132 str_value = str(value).lower().strip() 

133 

134 if str_value in self.TRUE_VALUES: 

135 return True 

136 if str_value in self.FALSE_VALUES: 

137 return False 

138 

139 # Default for ambiguous values 

140 return bool(metadata.default) 

141 

142 

143class ParseDatePipe(PipeProtocol): 

144 """PipeProtocol that parses string values to dates/datetimes. 

145 

146 Example: 

147 ```python 

148 class ReportController(Controller): 

149 @get("/reports") 

150 async def get_reports( 

151 self, 

152 @query(pipe=ParseDatePipe()) start_date: datetime 

153 ): 

154 ... 

155 ``` 

156 """ 

157 

158 async def transform(self, value: Any, metadata: ParamMetadata) -> datetime: 

159 """Parse value to datetime. 

160 

161 Args: 

162 value: The value to parse. 

163 metadata: Metadata about the parameter. 

164 

165 Returns: 

166 Parsed datetime. 

167 

168 Raises: 

169 ValueError: If value cannot be parsed as date. 

170 """ 

171 if value is None: 

172 if metadata.default is not None: 

173 default = metadata.default 

174 if isinstance(default, datetime): 

175 return default 

176 return datetime.fromisoformat(str(default)) 

177 raise ValueError(f"Missing required parameter: {metadata.name}") 

178 

179 if isinstance(value, datetime): 

180 return value 

181 

182 # Try ISO format first 

183 try: 

184 return datetime.fromisoformat(str(value)) 

185 except ValueError: 

186 pass 

187 

188 # Try other common formats 

189 formats = [ 

190 "%Y-%m-%d", 

191 "%Y-%m-%d %H:%M:%S", 

192 "%Y-%m-%dT%H:%M:%S", 

193 "%Y/%m/%d", 

194 ] 

195 

196 for fmt in formats: 

197 try: 

198 return datetime.strptime(str(value), fmt) 

199 except ValueError: 

200 continue 

201 

202 raise ValueError(f"Invalid date for {metadata.name}: {value}") 

203 

204 

205class DefaultValuePipe(PipeProtocol): 

206 """PipeProtocol that applies default values to missing parameters. 

207 

208 Example: 

209 ```python 

210 class SearchController(Controller): 

211 @get("/search") 

212 async def search( 

213 self, 

214 @query(pipe=DefaultValuePipe(default="")) query: str, 

215 @query(pipe=DefaultValuePipe(default=10)) limit: int, 

216 ): 

217 ... 

218 ``` 

219 """ 

220 

221 def __init__(self, default: Any = None): 

222 """Initialize with default value. 

223 

224 Args: 

225 default: The default value to apply. 

226 """ 

227 self._default = default 

228 

229 async def transform(self, value: Any, metadata: ParamMetadata) -> Any: 

230 """Apply default value. 

231 

232 Args: 

233 value: The value to check. 

234 metadata: Metadata about the parameter. 

235 

236 Returns: 

237 Value or default. 

238 """ 

239 if value is None: 

240 return self._default 

241 return value 

242 

243 

244__all__ = [ 

245 "DefaultValuePipe", 

246 "ParseBoolPipe", 

247 "ParseDatePipe", 

248 "ParseIntPipe", 

249 "ParseUUIDPipe", 

250]