Coverage for pydantic/experimental/pipeline.py: 92.34%

1"""Experimental pipeline API functionality. Be careful with this API, it's subject to change.""" 

2 

3from __future__ import annotations 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

4 

5import datetime 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

6import operator 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

7import re 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

8import sys 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

9from collections import deque 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

10from collections.abc import Container 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

11from dataclasses import dataclass 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

12from decimal import Decimal 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

13from functools import cached_property, partial 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

14from re import Pattern 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

15from typing import TYPE_CHECKING, Annotated, Any, Callable, Generic, Protocol, TypeVar, Union, overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

16 

17import annotated_types 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

18 

19if TYPE_CHECKING: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

20 from pydantic_core import core_schema as cs 

21 

22 from pydantic import GetCoreSchemaHandler 

23 

24from pydantic._internal._internal_dataclass import slots_true as _slots_true 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

25 

26if sys.version_info < (3, 10): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

27 EllipsisType = type(Ellipsis) 7 ctx1owexypz

28else: 

29 from types import EllipsisType 34 ctx1qArBagbhsCNfDEFGijklHIJtKuLcmdnvMO

30 

31__all__ = ['validate_as', 'validate_as_deferred', 'transform'] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

32 

33_slots_frozen = {**_slots_true, 'frozen': True} 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

34 

35 

36@dataclass(**_slots_frozen) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

37class _ValidateAs: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

38 tp: type[Any] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

39 strict: bool = False 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

40 

41 

42@dataclass 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

43class _ValidateAsDefer: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

44 func: Callable[[], type[Any]] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

45 

46 @cached_property 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

47 def tp(self) -> type[Any]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

48 return self.func() 

49 

50 

51@dataclass(**_slots_frozen) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

52class _Transform: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

53 func: Callable[[Any], Any] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

54 

55 

56@dataclass(**_slots_frozen) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

57class _PipelineOr: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

58 left: _Pipeline[Any, Any] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

59 right: _Pipeline[Any, Any] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

60 

61 

62@dataclass(**_slots_frozen) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

63class _PipelineAnd: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

64 left: _Pipeline[Any, Any] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

65 right: _Pipeline[Any, Any] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

66 

67 

68@dataclass(**_slots_frozen) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

69class _Eq: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

70 value: Any 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

71 

72 

73@dataclass(**_slots_frozen) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

74class _NotEq: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

75 value: Any 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

76 

77 

78@dataclass(**_slots_frozen) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

79class _In: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

80 values: Container[Any] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

81 

82 

83@dataclass(**_slots_frozen) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

84class _NotIn: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

85 values: Container[Any] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

86 

87 

88_ConstraintAnnotation = Union[ 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

89 annotated_types.Le, 

90 annotated_types.Ge, 

91 annotated_types.Lt, 

92 annotated_types.Gt, 

93 annotated_types.Len, 

94 annotated_types.MultipleOf, 

95 annotated_types.Timezone, 

96 annotated_types.Interval, 

97 annotated_types.Predicate, 

98 # common predicates not included in annotated_types 

99 _Eq, 

100 _NotEq, 

101 _In, 

102 _NotIn, 

103 # regular expressions 

104 Pattern[str], 

105] 

106 

107 

108@dataclass(**_slots_frozen) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

109class _Constraint: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

110 constraint: _ConstraintAnnotation 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

111 

112 

113_Step = Union[_ValidateAs, _ValidateAsDefer, _Transform, _PipelineOr, _PipelineAnd, _Constraint] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

114 

115_InT = TypeVar('_InT') 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

116_OutT = TypeVar('_OutT') 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

117_NewOutT = TypeVar('_NewOutT') 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

118 

119 

120class _FieldTypeMarker: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

121 pass 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

122 

123 

124# TODO: ultimately, make this public, see https://github.com/pydantic/pydantic/pull/9459#discussion_r1628197626 

125# Also, make this frozen eventually, but that doesn't work right now because of the generic base 

126# Which attempts to modify __orig_base__ and such. 

127# We could go with a manual freeze, but that seems overkill for now. 

128@dataclass(**_slots_true) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

129class _Pipeline(Generic[_InT, _OutT]): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

130 """Abstract representation of a chain of validation, transformation, and parsing steps.""" 

131 

132 _steps: tuple[_Step, ...] 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

133 

134 def transform( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

135 self, 

136 func: Callable[[_OutT], _NewOutT], 

137 ) -> _Pipeline[_InT, _NewOutT]: 

138 """Transform the output of the previous step. 

139 

140 If used as the first step in a pipeline, the type of the field is used. 

141 That is, the transformation is applied to after the value is parsed to the field's type. 

142 """ 

143 return _Pipeline[_InT, _NewOutT](self._steps + (_Transform(func),)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

144 

145 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

146 def validate_as(self, tp: type[_NewOutT], *, strict: bool = ...) -> _Pipeline[_InT, _NewOutT]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

147 

148 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

149 def validate_as(self, tp: EllipsisType, *, strict: bool = ...) -> _Pipeline[_InT, Any]: # type: ignore 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

150 ... 

151 

152 def validate_as(self, tp: type[_NewOutT] | EllipsisType, *, strict: bool = False) -> _Pipeline[_InT, Any]: # type: ignore 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

153 """Validate / parse the input into a new type. 

154 

155 If no type is provided, the type of the field is used. 

156 

157 Types are parsed in Pydantic's `lax` mode by default, 

158 but you can enable `strict` mode by passing `strict=True`. 

159 """ 

160 if isinstance(tp, EllipsisType): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

161 return _Pipeline[_InT, Any](self._steps + (_ValidateAs(_FieldTypeMarker, strict=strict),)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

162 return _Pipeline[_InT, _NewOutT](self._steps + (_ValidateAs(tp, strict=strict),)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

163 

164 def validate_as_deferred(self, func: Callable[[], type[_NewOutT]]) -> _Pipeline[_InT, _NewOutT]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

165 """Parse the input into a new type, deferring resolution of the type until the current class 

166 is fully defined. 

167 

168 This is useful when you need to reference the class in it's own type annotations. 

169 """ 

170 return _Pipeline[_InT, _NewOutT](self._steps + (_ValidateAsDefer(func),)) 

171 

172 # constraints 

173 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

174 def constrain(self: _Pipeline[_InT, _NewOutGe], constraint: annotated_types.Ge) -> _Pipeline[_InT, _NewOutGe]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

175 

176 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

177 def constrain(self: _Pipeline[_InT, _NewOutGt], constraint: annotated_types.Gt) -> _Pipeline[_InT, _NewOutGt]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

178 

179 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

180 def constrain(self: _Pipeline[_InT, _NewOutLe], constraint: annotated_types.Le) -> _Pipeline[_InT, _NewOutLe]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

181 

182 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

183 def constrain(self: _Pipeline[_InT, _NewOutLt], constraint: annotated_types.Lt) -> _Pipeline[_InT, _NewOutLt]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

184 

185 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

186 def constrain( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

187 self: _Pipeline[_InT, _NewOutLen], constraint: annotated_types.Len 14 ctx1agbhefijklcmdn

188 ) -> _Pipeline[_InT, _NewOutLen]: ... 14 ctx1agbhefijklcmdn

189 

190 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

191 def constrain( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

192 self: _Pipeline[_InT, _NewOutT], constraint: annotated_types.MultipleOf 14 ctx1agbhefijklcmdn

193 ) -> _Pipeline[_InT, _NewOutT]: ... 14 ctx1agbhefijklcmdn

194 

195 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

196 def constrain( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

197 self: _Pipeline[_InT, _NewOutDatetime], constraint: annotated_types.Timezone 14 ctx1agbhefijklcmdn

198 ) -> _Pipeline[_InT, _NewOutDatetime]: ... 14 ctx1agbhefijklcmdn

199 

200 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

201 def constrain(self: _Pipeline[_InT, _OutT], constraint: annotated_types.Predicate) -> _Pipeline[_InT, _OutT]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

202 

203 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

204 def constrain( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

205 self: _Pipeline[_InT, _NewOutInterval], constraint: annotated_types.Interval 14 ctx1agbhefijklcmdn

206 ) -> _Pipeline[_InT, _NewOutInterval]: ... 14 ctx1agbhefijklcmdn

207 

208 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

209 def constrain(self: _Pipeline[_InT, _OutT], constraint: _Eq) -> _Pipeline[_InT, _OutT]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

210 

211 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

212 def constrain(self: _Pipeline[_InT, _OutT], constraint: _NotEq) -> _Pipeline[_InT, _OutT]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

213 

214 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

215 def constrain(self: _Pipeline[_InT, _OutT], constraint: _In) -> _Pipeline[_InT, _OutT]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

216 

217 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

218 def constrain(self: _Pipeline[_InT, _OutT], constraint: _NotIn) -> _Pipeline[_InT, _OutT]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

219 

220 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

221 def constrain(self: _Pipeline[_InT, _NewOutT], constraint: Pattern[str]) -> _Pipeline[_InT, _NewOutT]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

222 

223 def constrain(self, constraint: _ConstraintAnnotation) -> Any: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

224 """Constrain a value to meet a certain condition. 

225 

226 We support most conditions from `annotated_types`, as well as regular expressions. 

227 

228 Most of the time you'll be calling a shortcut method like `gt`, `lt`, `len`, etc 

229 so you don't need to call this directly. 

230 """ 

231 return _Pipeline[_InT, _OutT](self._steps + (_Constraint(constraint),)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

232 

233 def predicate(self: _Pipeline[_InT, _NewOutT], func: Callable[[_NewOutT], bool]) -> _Pipeline[_InT, _NewOutT]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

234 """Constrain a value to meet a certain predicate.""" 

235 return self.constrain(annotated_types.Predicate(func)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

236 

237 def gt(self: _Pipeline[_InT, _NewOutGt], gt: _NewOutGt) -> _Pipeline[_InT, _NewOutGt]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

238 """Constrain a value to be greater than a certain value.""" 

239 return self.constrain(annotated_types.Gt(gt)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

240 

241 def lt(self: _Pipeline[_InT, _NewOutLt], lt: _NewOutLt) -> _Pipeline[_InT, _NewOutLt]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

242 """Constrain a value to be less than a certain value.""" 

243 return self.constrain(annotated_types.Lt(lt)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

244 

245 def ge(self: _Pipeline[_InT, _NewOutGe], ge: _NewOutGe) -> _Pipeline[_InT, _NewOutGe]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

246 """Constrain a value to be greater than or equal to a certain value.""" 

247 return self.constrain(annotated_types.Ge(ge)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

248 

249 def le(self: _Pipeline[_InT, _NewOutLe], le: _NewOutLe) -> _Pipeline[_InT, _NewOutLe]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

250 """Constrain a value to be less than or equal to a certain value.""" 

251 return self.constrain(annotated_types.Le(le)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

252 

253 def len(self: _Pipeline[_InT, _NewOutLen], min_len: int, max_len: int | None = None) -> _Pipeline[_InT, _NewOutLen]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

254 """Constrain a value to have a certain length.""" 

255 return self.constrain(annotated_types.Len(min_len, max_len)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

256 

257 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

258 def multiple_of(self: _Pipeline[_InT, _NewOutDiv], multiple_of: _NewOutDiv) -> _Pipeline[_InT, _NewOutDiv]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

259 

260 @overload 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

261 def multiple_of(self: _Pipeline[_InT, _NewOutMod], multiple_of: _NewOutMod) -> _Pipeline[_InT, _NewOutMod]: ... 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

262 

263 def multiple_of(self: _Pipeline[_InT, Any], multiple_of: Any) -> _Pipeline[_InT, Any]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

264 """Constrain a value to be a multiple of a certain number.""" 

265 return self.constrain(annotated_types.MultipleOf(multiple_of)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

266 

267 def eq(self: _Pipeline[_InT, _OutT], value: _OutT) -> _Pipeline[_InT, _OutT]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

268 """Constrain a value to be equal to a certain value.""" 

269 return self.constrain(_Eq(value)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

270 

271 def not_eq(self: _Pipeline[_InT, _OutT], value: _OutT) -> _Pipeline[_InT, _OutT]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

272 """Constrain a value to not be equal to a certain value.""" 

273 return self.constrain(_NotEq(value)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

274 

275 def in_(self: _Pipeline[_InT, _OutT], values: Container[_OutT]) -> _Pipeline[_InT, _OutT]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

276 """Constrain a value to be in a certain set.""" 

277 return self.constrain(_In(values)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

278 

279 def not_in(self: _Pipeline[_InT, _OutT], values: Container[_OutT]) -> _Pipeline[_InT, _OutT]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

280 """Constrain a value to not be in a certain set.""" 

281 return self.constrain(_NotIn(values)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

282 

283 # timezone methods 

284 def datetime_tz_naive(self: _Pipeline[_InT, datetime.datetime]) -> _Pipeline[_InT, datetime.datetime]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

285 return self.constrain(annotated_types.Timezone(None)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

286 

287 def datetime_tz_aware(self: _Pipeline[_InT, datetime.datetime]) -> _Pipeline[_InT, datetime.datetime]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

288 return self.constrain(annotated_types.Timezone(...)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

289 

290 def datetime_tz( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

291 self: _Pipeline[_InT, datetime.datetime], tz: datetime.tzinfo 

292 ) -> _Pipeline[_InT, datetime.datetime]: 

293 return self.constrain(annotated_types.Timezone(tz)) # type: ignore 

294 

295 def datetime_with_tz( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

296 self: _Pipeline[_InT, datetime.datetime], tz: datetime.tzinfo | None 

297 ) -> _Pipeline[_InT, datetime.datetime]: 

298 return self.transform(partial(datetime.datetime.replace, tzinfo=tz)) 

299 

300 # string methods 

301 def str_lower(self: _Pipeline[_InT, str]) -> _Pipeline[_InT, str]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

302 return self.transform(str.lower) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

303 

304 def str_upper(self: _Pipeline[_InT, str]) -> _Pipeline[_InT, str]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

305 return self.transform(str.upper) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

306 

307 def str_title(self: _Pipeline[_InT, str]) -> _Pipeline[_InT, str]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

308 return self.transform(str.title) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

309 

310 def str_strip(self: _Pipeline[_InT, str]) -> _Pipeline[_InT, str]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

311 return self.transform(str.strip) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

312 

313 def str_pattern(self: _Pipeline[_InT, str], pattern: str) -> _Pipeline[_InT, str]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

314 return self.constrain(re.compile(pattern)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

315 

316 def str_contains(self: _Pipeline[_InT, str], substring: str) -> _Pipeline[_InT, str]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

317 return self.predicate(lambda v: substring in v) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

318 

319 def str_starts_with(self: _Pipeline[_InT, str], prefix: str) -> _Pipeline[_InT, str]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

320 return self.predicate(lambda v: v.startswith(prefix)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

321 

322 def str_ends_with(self: _Pipeline[_InT, str], suffix: str) -> _Pipeline[_InT, str]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

323 return self.predicate(lambda v: v.endswith(suffix)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

324 

325 # operators 

326 def otherwise(self, other: _Pipeline[_OtherIn, _OtherOut]) -> _Pipeline[_InT | _OtherIn, _OutT | _OtherOut]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

327 """Combine two validation chains, returning the result of the first chain if it succeeds, and the second chain if it fails.""" 

328 return _Pipeline((_PipelineOr(self, other),)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

329 

330 __or__ = otherwise 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

331 

332 def then(self, other: _Pipeline[_OutT, _OtherOut]) -> _Pipeline[_InT, _OtherOut]: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

333 """Pipe the result of one validation chain into another.""" 

334 return _Pipeline((_PipelineAnd(self, other),)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

335 

336 __and__ = then 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

337 

338 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> cs.CoreSchema: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

339 from pydantic_core import core_schema as cs 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

340 

341 queue = deque(self._steps) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

342 

343 s = None 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

344 

345 while queue: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

346 step = queue.popleft() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

347 s = _apply_step(step, s, handler, source_type) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

348 

349 s = s or cs.any_schema() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

350 return s 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

351 

352 def __supports_type__(self, _: _OutT) -> bool: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

353 raise NotImplementedError 

354 

355 

356validate_as = _Pipeline[Any, Any](()).validate_as 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

357validate_as_deferred = _Pipeline[Any, Any](()).validate_as_deferred 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

358transform = _Pipeline[Any, Any]((_ValidateAs(_FieldTypeMarker),)).transform 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

359 

360 

361def _check_func( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

362 func: Callable[[Any], bool], predicate_err: str | Callable[[], str], s: cs.CoreSchema | None 

363) -> cs.CoreSchema: 

364 from pydantic_core import core_schema as cs 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

365 

366 def handler(v: Any) -> Any: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

367 if func(v): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

368 return v 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

369 raise ValueError(f'Expected {predicate_err if isinstance(predicate_err, str) else predicate_err()}') 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

370 

371 if s is None: 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

372 return cs.no_info_plain_validator_function(handler) 

373 else: 

374 return cs.no_info_after_validator_function(handler, s) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

375 

376 

377def _apply_step(step: _Step, s: cs.CoreSchema | None, handler: GetCoreSchemaHandler, source_type: Any) -> cs.CoreSchema: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

378 from pydantic_core import core_schema as cs 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

379 

380 if isinstance(step, _ValidateAs): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

381 s = _apply_parse(s, step.tp, step.strict, handler, source_type) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

382 elif isinstance(step, _ValidateAsDefer): 382 ↛ 383line 382 didn't jump to line 383 because the condition on line 382 was never true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

383 s = _apply_parse(s, step.tp, False, handler, source_type) 

384 elif isinstance(step, _Transform): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

385 s = _apply_transform(s, step.func, handler) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

386 elif isinstance(step, _Constraint): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

387 s = _apply_constraint(s, step.constraint) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

388 elif isinstance(step, _PipelineOr): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

389 s = cs.union_schema([handler(step.left), handler(step.right)]) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

390 else: 

391 assert isinstance(step, _PipelineAnd) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

392 s = cs.chain_schema([handler(step.left), handler(step.right)]) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

393 return s 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

394 

395 

396def _apply_parse( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

397 s: cs.CoreSchema | None, 

398 tp: type[Any], 

399 strict: bool, 

400 handler: GetCoreSchemaHandler, 

401 source_type: Any, 

402) -> cs.CoreSchema: 

403 from pydantic_core import core_schema as cs 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

404 

405 from pydantic import Strict 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

406 

407 if tp is _FieldTypeMarker: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

408 return cs.chain_schema([s, handler(source_type)]) if s else handler(source_type) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

409 

410 if strict: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

411 tp = Annotated[tp, Strict()] # type: ignore 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

412 

413 if s and s['type'] == 'any': 413 ↛ 414line 413 didn't jump to line 414 because the condition on line 413 was never true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

414 return handler(tp) 

415 else: 

416 return cs.chain_schema([s, handler(tp)]) if s else handler(tp) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

417 

418 

419def _apply_transform( 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

420 s: cs.CoreSchema | None, func: Callable[[Any], Any], handler: GetCoreSchemaHandler 

421) -> cs.CoreSchema: 

422 from pydantic_core import core_schema as cs 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

423 

424 if s is None: 424 ↛ 425line 424 didn't jump to line 425 because the condition on line 424 was never true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

425 return cs.no_info_plain_validator_function(func) 

426 

427 if s['type'] == 'str': 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

428 if func is str.strip: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

429 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

430 s['strip_whitespace'] = True 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

431 return s 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

432 elif func is str.lower: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

433 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

434 s['to_lower'] = True 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

435 return s 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

436 elif func is str.upper: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

437 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

438 s['to_upper'] = True 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

439 return s 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

440 

441 return cs.no_info_after_validator_function(func, s) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

442 

443 

444def _apply_constraint( # noqa: C901 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

445 s: cs.CoreSchema | None, constraint: _ConstraintAnnotation 

446) -> cs.CoreSchema: 

447 """Apply a single constraint to a schema.""" 

448 if isinstance(constraint, annotated_types.Gt): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

449 gt = constraint.gt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

450 if s and s['type'] in {'int', 'float', 'decimal'}: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

451 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

452 if s['type'] == 'int' and isinstance(gt, int): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

453 s['gt'] = gt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

454 elif s['type'] == 'float' and isinstance(gt, float): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

455 s['gt'] = gt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

456 elif s['type'] == 'decimal' and isinstance(gt, Decimal): 456 ↛ 646line 456 didn't jump to line 646 because the condition on line 456 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

457 s['gt'] = gt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

458 else: 

459 

460 def check_gt(v: Any) -> bool: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

461 return v > gt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

462 

463 s = _check_func(check_gt, f'> {gt}', s) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

464 elif isinstance(constraint, annotated_types.Ge): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

465 ge = constraint.ge 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

466 if s and s['type'] in {'int', 'float', 'decimal'}: 466 ↛ 475line 466 didn't jump to line 475 because the condition on line 466 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

467 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

468 if s['type'] == 'int' and isinstance(ge, int): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

469 s['ge'] = ge 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

470 elif s['type'] == 'float' and isinstance(ge, float): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

471 s['ge'] = ge 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

472 elif s['type'] == 'decimal' and isinstance(ge, Decimal): 472 ↛ 475line 472 didn't jump to line 475 because the condition on line 472 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

473 s['ge'] = ge 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

474 

475 def check_ge(v: Any) -> bool: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

476 return v >= ge 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

477 

478 s = _check_func(check_ge, f'>= {ge}', s) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

479 elif isinstance(constraint, annotated_types.Lt): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

480 lt = constraint.lt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

481 if s and s['type'] in {'int', 'float', 'decimal'}: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

482 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

483 if s['type'] == 'int' and isinstance(lt, int): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

484 s['lt'] = lt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

485 elif s['type'] == 'float' and isinstance(lt, float): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

486 s['lt'] = lt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

487 elif s['type'] == 'decimal' and isinstance(lt, Decimal): 487 ↛ 490line 487 didn't jump to line 490 because the condition on line 487 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

488 s['lt'] = lt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

489 

490 def check_lt(v: Any) -> bool: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

491 return v < lt 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

492 

493 s = _check_func(check_lt, f'< {lt}', s) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

494 elif isinstance(constraint, annotated_types.Le): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

495 le = constraint.le 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

496 if s and s['type'] in {'int', 'float', 'decimal'}: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

497 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

498 if s['type'] == 'int' and isinstance(le, int): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

499 s['le'] = le 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

500 elif s['type'] == 'float' and isinstance(le, float): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

501 s['le'] = le 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

502 elif s['type'] == 'decimal' and isinstance(le, Decimal): 502 ↛ 505line 502 didn't jump to line 505 because the condition on line 502 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

503 s['le'] = le 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

504 

505 def check_le(v: Any) -> bool: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

506 return v <= le 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

507 

508 s = _check_func(check_le, f'<= {le}', s) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

509 elif isinstance(constraint, annotated_types.Len): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

510 min_len = constraint.min_length 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

511 max_len = constraint.max_length 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

512 

513 if s and s['type'] in {'str', 'list', 'tuple', 'set', 'frozenset', 'dict'}: 513 ↛ 528line 513 didn't jump to line 528 because the condition on line 513 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

514 assert ( 8 ctx1owefxypz

515 s['type'] == 'str' 

516 or s['type'] == 'list' 

517 or s['type'] == 'tuple' 

518 or s['type'] == 'set' 

519 or s['type'] == 'dict' 

520 or s['type'] == 'frozenset' 

521 ) 

522 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

523 if min_len != 0: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

524 s['min_length'] = min_len 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

525 if max_len is not None: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

526 s['max_length'] = max_len 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

527 

528 def check_len(v: Any) -> bool: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

529 if max_len is not None: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

530 return (min_len <= len(v)) and (len(v) <= max_len) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

531 return min_len <= len(v) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

532 

533 s = _check_func(check_len, f'length >= {min_len} and length <= {max_len}', s) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

534 elif isinstance(constraint, annotated_types.MultipleOf): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

535 multiple_of = constraint.multiple_of 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

536 if s and s['type'] in {'int', 'float', 'decimal'}: 536 ↛ 545line 536 didn't jump to line 545 because the condition on line 536 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

537 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

538 if s['type'] == 'int' and isinstance(multiple_of, int): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

539 s['multiple_of'] = multiple_of 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

540 elif s['type'] == 'float' and isinstance(multiple_of, float): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

541 s['multiple_of'] = multiple_of 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

542 elif s['type'] == 'decimal' and isinstance(multiple_of, Decimal): 542 ↛ 545line 542 didn't jump to line 545 because the condition on line 542 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

543 s['multiple_of'] = multiple_of 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

544 

545 def check_multiple_of(v: Any) -> bool: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

546 return v % multiple_of == 0 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

547 

548 s = _check_func(check_multiple_of, f'% {multiple_of} == 0', s) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

549 elif isinstance(constraint, annotated_types.Timezone): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

550 tz = constraint.tz 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

551 

552 if tz is ...: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

553 if s and s['type'] == 'datetime': 553 ↛ 558line 553 didn't jump to line 558 because the condition on line 553 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

554 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

555 s['tz_constraint'] = 'aware' 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

556 else: 

557 

558 def check_tz_aware(v: object) -> bool: 

559 assert isinstance(v, datetime.datetime) 

560 return v.tzinfo is not None 

561 

562 s = _check_func(check_tz_aware, 'timezone aware', s) 

563 elif tz is None: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

564 if s and s['type'] == 'datetime': 564 ↛ 569line 564 didn't jump to line 569 because the condition on line 564 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

565 s = s.copy() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

566 s['tz_constraint'] = 'naive' 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

567 else: 

568 

569 def check_tz_naive(v: object) -> bool: 

570 assert isinstance(v, datetime.datetime) 

571 return v.tzinfo is None 

572 

573 s = _check_func(check_tz_naive, 'timezone naive', s) 

574 else: 

575 raise NotImplementedError('Constraining to a specific timezone is not yet supported') 

576 elif isinstance(constraint, annotated_types.Interval): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

577 if constraint.ge: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

578 s = _apply_constraint(s, annotated_types.Ge(constraint.ge)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

579 if constraint.gt: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

580 s = _apply_constraint(s, annotated_types.Gt(constraint.gt)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

581 if constraint.le: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

582 s = _apply_constraint(s, annotated_types.Le(constraint.le)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

583 if constraint.lt: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

584 s = _apply_constraint(s, annotated_types.Lt(constraint.lt)) 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

585 assert s is not None 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

586 elif isinstance(constraint, annotated_types.Predicate): 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

587 func = constraint.func 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

588 

589 if func.__name__ == '<lambda>': 589 ↛ 605line 589 didn't jump to line 605 because the condition on line 589 was always true41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

590 # attempt to extract the source code for a lambda function 

591 # to use as the function name in error messages 

592 # TODO: is there a better way? should we just not do this? 

593 import inspect 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

594 

595 try: 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

596 source = inspect.getsource(func).strip() 41 ctx1owqArBagbhsCNefxyDEFGijklHIJpztKuLcmdnvMO

597 source = source.removesuffix(')') 39 ctx1owqArBagbhsCefxyDEFGijklHIJpztKuLcmdnvM

598 lambda_source_code = '`' + ''.join(''.join(source.split('lambda ')[1:]).split(':')[1:]).strip() + '`' 39 ctx1owqArBagbhsCefxyDEFGijklHIJpztKuLcmdnvM