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
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Parse pipes for type conversion.
3Pipes that parse string values into specific types.
4"""
6from __future__ import annotations
8from datetime import datetime
9from typing import Any
10import uuid
12from lexigram.web.protocols import ParamMetadata, PipeProtocol
15class ParseIntPipe(PipeProtocol):
16 """PipeProtocol that parses string values to integers.
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 """
27 def __init__(self, strict: bool = True):
28 """Initialize the pipe.
30 Args:
31 strict: If True, raises error on invalid input. If False, returns default.
32 """
33 self._strict = strict
35 async def transform(self, value: Any, metadata: ParamMetadata) -> int:
36 """Parse value to integer.
38 Args:
39 value: The value to parse.
40 metadata: Metadata about the parameter.
42 Returns:
43 Parsed integer.
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
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
63class ParseUUIDPipe(PipeProtocol):
64 """PipeProtocol that parses string values to UUIDs.
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 """
75 async def transform(self, value: Any, metadata: ParamMetadata) -> uuid.UUID:
76 """Parse value to UUID.
78 Args:
79 value: The value to parse.
80 metadata: Metadata about the parameter.
82 Returns:
83 Parsed UUID.
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}")
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
99class ParseBoolPipe(PipeProtocol):
100 """PipeProtocol that parses string values to booleans.
102 Accepts: "true", "false", "1", "0", "yes", "no"
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 """
113 TRUE_VALUES = {"true", "1", "yes", "on"}
114 FALSE_VALUES = {"false", "0", "no", "off"}
116 async def transform(self, value: Any, metadata: ParamMetadata) -> bool:
117 """Parse value to boolean.
119 Args:
120 value: The value to parse.
121 metadata: Metadata about the parameter.
123 Returns:
124 Parsed boolean.
125 """
126 if value is None:
127 return metadata.default if metadata.default is not None else True
129 if isinstance(value, bool):
130 return value
132 str_value = str(value).lower().strip()
134 if str_value in self.TRUE_VALUES:
135 return True
136 if str_value in self.FALSE_VALUES:
137 return False
139 # Default for ambiguous values
140 return bool(metadata.default)
143class ParseDatePipe(PipeProtocol):
144 """PipeProtocol that parses string values to dates/datetimes.
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 """
158 async def transform(self, value: Any, metadata: ParamMetadata) -> datetime:
159 """Parse value to datetime.
161 Args:
162 value: The value to parse.
163 metadata: Metadata about the parameter.
165 Returns:
166 Parsed datetime.
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}")
179 if isinstance(value, datetime):
180 return value
182 # Try ISO format first
183 try:
184 return datetime.fromisoformat(str(value))
185 except ValueError:
186 pass
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 ]
196 for fmt in formats:
197 try:
198 return datetime.strptime(str(value), fmt)
199 except ValueError:
200 continue
202 raise ValueError(f"Invalid date for {metadata.name}: {value}")
205class DefaultValuePipe(PipeProtocol):
206 """PipeProtocol that applies default values to missing parameters.
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 """
221 def __init__(self, default: Any = None):
222 """Initialize with default value.
224 Args:
225 default: The default value to apply.
226 """
227 self._default = default
229 async def transform(self, value: Any, metadata: ParamMetadata) -> Any:
230 """Apply default value.
232 Args:
233 value: The value to check.
234 metadata: Metadata about the parameter.
236 Returns:
237 Value or default.
238 """
239 if value is None:
240 return self._default
241 return value
244__all__ = [
245 "DefaultValuePipe",
246 "ParseBoolPipe",
247 "ParseDatePipe",
248 "ParseIntPipe",
249 "ParseUUIDPipe",
250]