Coverage for an_website/utils/data_parsing.py: 60.714%

1# This program is free software: you can redistribute it and/or modify 

2# it under the terms of the GNU Affero General Public License as 

3# published by the Free Software Foundation, either version 3 of the 

4# License, or (at your option) any later version. 

5# 

6# This program is distributed in the hope that it will be useful, 

7# but WITHOUT ANY WARRANTY; without even the implied warranty of 

8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

9# GNU Affero General Public License for more details. 

10# 

11# You should have received a copy of the GNU Affero General Public License 

12# along with this program. If not, see <https://www.gnu.org/licenses/>. 

13 

14"""Parse data into classes.""" 

15 

16from __future__ import annotations 

17 

18import contextlib 

19import functools 

20import inspect 

21import typing 

22from collections.abc import Callable, Iterable, Mapping 

23from inspect import Parameter 

24from types import UnionType 

25from typing import Any, Literal, TypeVar, get_origin 

26 

27from tornado.web import HTTPError, RequestHandler 

28 

29from .utils import str_to_bool 

30 

31T = TypeVar("T") 

32 

33 

34def parse( # noqa: C901 # pylint: disable=too-many-branches 

35 type_: type[T], 

36 data: Iterable[Mapping[str, Any]] | Mapping[str, Any] | Any, 

37 *, 

38 strict: bool = False, 

39) -> T: 

40 """Parse a data.""" 

41 # pylint: disable=too-many-return-statements, too-complex 

42 if data is None and type_ is None: 

43 return None # type: ignore[unreachable] 

44 

45 simple_type: type = get_origin(type_) or type_ 

46 

47 if ( 

48 simple_type is type_ 

49 and not isinstance(type_, str) # type: ignore[redundant-expr] 

50 and isinstance(data, type_) 

51 ): 

52 return data 

53 

54 if simple_type is Literal: # type: ignore[comparison-overlap] 

55 possible = tuple(typing.get_args(type_)) 

56 for pos in possible: 

57 if pos == data: 

58 return typing.cast(T, pos) 

59 if isinstance(data, str): 

60 data = data.strip() 

61 for pos in possible: 

62 if str(pos) == data: 

63 return typing.cast(T, pos) 

64 raise ValueError(f"Unable to parse {data!r} into {type_}") 

65 

66 if simple_type == UnionType: 

67 possible = tuple(typing.get_args(type_)) 

68 for pos in possible: 

69 with contextlib.suppress(ValueError): 

70 return typing.cast(T, parse(pos, data, strict=strict)) 

71 raise ValueError(f"Unable to parse {data!r} into {type_}") 

72 

73 if simple_type in {list, "list"} and isinstance(data, list): 

74 return _parse_list(type_, data, strict=strict) 

75 if type_ in {bool, "bool"}: 

76 return typing.cast(T, _parse_bool(data, strict=strict)) 

77 if type_ in {str, "str"}: 

78 return typing.cast(T, _parse_str(data, strict=strict)) 

79 if type_ in {int, "int"}: 

80 return typing.cast(T, _parse_int(data, strict=strict)) 

81 if type_ in {float, "float"}: 

82 return typing.cast(T, _parse_float(data, strict=strict)) 

83 if hasattr(type_, "__init__") and isinstance(data, dict): 

84 return _parse_class(type_, data, strict=strict) 

85 

86 raise ValueError(f"Unable to parse {data!r} into {type_}") 

87 

88 

89def _parse_str(data: Any, *, strict: bool) -> str: 

90 """Parse data into str.""" 

91 if isinstance(data, str): 

92 return data 

93 if strict: 

94 raise ValueError(f"{data!r} is not str.") 

95 return str(data) 

96 

97 

98def _parse_bool(data: Any, *, strict: bool) -> bool: 

99 """Parse data into bool.""" 

100 if isinstance(data, bool): 

101 return data 

102 if strict: 

103 raise ValueError(f"{data!r} is not bool.") 

104 if data in {0, 1}: 

105 return bool(data) 

106 if isinstance(data, str): 

107 return str_to_bool(data) 

108 raise ValueError(f"Cannot parse {data!r} into bool.") 

109 

110 

111def _parse_int(data: Any, *, strict: bool) -> int: 

112 """Parse data into int.""" 

113 if isinstance(data, int): 

114 return data 

115 if isinstance(data, float) and int(data) == data: 

116 return int(data) 

117 if strict: 

118 raise ValueError(f"{data!r} is not a number.") 

119 if isinstance(data, str): 

120 return int(data, base=0) 

121 if isinstance(data, bool): 

122 return int(data) 

123 raise ValueError(f"Cannot parse {data!r} into int.") 

124 

125 

126def _parse_float(data: Any, *, strict: bool) -> float: 

127 """Parse data into float.""" 

128 if isinstance(data, float): 

129 return data 

130 if isinstance(data, int): 

131 return float(data) 

132 if strict: 

133 raise ValueError(f"{data!r} is not a number.") 

134 if isinstance(data, str): 

135 return int(data) 

136 if isinstance(data, bool): 

137 return int(data) 

138 raise ValueError(f"Cannot parse {data!r} into int.") 

139 

140 

141def _parse_list( 

142 type_: type[T], data: Iterable[Mapping[str, Any]], *, strict: bool 

143) -> T: 

144 """Parse a list of data.""" 

145 args = typing.get_args(type_) 

146 if len(args) != 1: 

147 raise ValueError(f"{type_=} should be list[...]") 

148 return typing.cast( 

149 T, [parse(args[0], spam, strict=strict) for spam in data] 

150 ) 

151 

152 

153def _parse_class(type_: type[T], data: Mapping[str, Any], *, strict: bool) -> T: 

154 """Parse data into a class.""" 

155 signature = inspect.signature(type_.__init__, eval_str=True) 

156 args: list[Any] = [] 

157 kwargs: dict[str, Any] = {} 

158 in_positional = False 

159 

160 def add(_name: str, _value: Any) -> None: 

161 if in_positional: 

162 args.append(_value) 

163 else: 

164 kwargs[_name] = _value 

165 

166 first = True 

167 for arg_name, param in signature.parameters.items(): 

168 if first: 

169 first = False 

170 continue 

171 if param.kind in {Parameter.VAR_KEYWORD, Parameter.KEYWORD_ONLY}: 

172 in_positional = True 

173 if arg_name not in data: 

174 if param.default == Parameter.empty: 

175 raise ValueError(f"Missing required argument {arg_name!r}") 

176 add(arg_name, param.default) 

177 continue 

178 value = data[arg_name] 

179 if param.annotation == Parameter.empty: 

180 if strict: 

181 raise ValueError(f"Missing type annotation for {arg_name!r}") 

182 add(arg_name, value) 

183 continue 

184 add(arg_name, parse(param.annotation, value, strict=strict)) 

185 

186 return type_(*args, **kwargs) 

187 

188 

189def parse_args( 

190 *, type_: Any, name: str = "args", validation_method: str | None = None 

191) -> Callable[[Callable[..., T]], Callable[..., T]]: 

192 """Insert kwarg with name and type into function.""" 

193 

194 def _inner(func: Callable[..., T]) -> Callable[..., T]: 

195 @functools.wraps(func) 

196 def new_func(self: RequestHandler, *args: Any, **kwargs: Any) -> T: 

197 arguments: dict[str, str] = {} 

198 for key, values in self.request.arguments.items(): 

199 if len(values) == 1: 

200 arguments[key] = values[0].decode("UTF-8", "replace") 

201 else: 

202 raise HTTPError( # we don't want to guess 

203 400, reason=f"Given multiple values for {key!r}" 

204 ) 

205 try: 

206 _data = parse(type_, arguments, strict=False) 

207 except ValueError as err: 

208 raise HTTPError(400, reason=err.args[0]) from err 

209 if validation_method: 

210 getattr(_data, validation_method)() 

211 return func(self, *args, **kwargs, **{name: _data}) 

212 

213 return new_func 

214 

215 return _inner