Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/_header_property.py: 44%

1from __future__ import annotations 

2 

3import typing as t 

4 

5from .datastructures.set import HeaderSet 

6from .http import dump_header 

7 

8if t.TYPE_CHECKING: 

9 from .sansio.response import Response 

10 

11T = t.TypeVar("T") 

12 

13 

14class environ_property(t.Generic[T]): 

15 """A property that returns a key from :attr:`.Request.environ`. 

16 

17 .. deprecated:: 3.2 

18 Will be removed in Werkzeug 4.0. Access ``environ`` directly instead. 

19 """ 

20 

21 def __init__( 

22 self, 

23 name: str, 

24 default: T | None = None, 

25 load_func: t.Callable[[str], T] | None = None, 

26 dump_func: t.Callable[[T], str] = str, 

27 read_only: bool = True, 

28 doc: str | None = None, 

29 ) -> None: 

30 self.name = name 

31 self.default = default 

32 self.load_func = load_func 

33 self.dump_func = dump_func 

34 self.read_only = read_only 

35 

36 if doc is not None: 

37 self.__doc__ = doc 

38 

39 @t.overload 

40 def __get__(self, obj: None, cls: type[t.Any], /) -> t.Self: ... 

41 @t.overload 

42 def __get__(self, obj: t.Any, cls: type[t.Any] | None = ..., /) -> T: ... 

43 def __get__( 

44 self, obj: t.Any | None = None, cls: type[t.Any] | None = None, / 

45 ) -> T | t.Self: 

46 if obj is None: 

47 return self 

48 

49 try: 

50 value: t.Any = obj.environ[self.name] 

51 except KeyError: 

52 value = self.default 

53 

54 if self.load_func is not None: 

55 try: 

56 value = self.load_func(value) 

57 except (ValueError, TypeError): 

58 value = self.default 

59 

60 return value # type: ignore[no-any-return] 

61 

62 def __set__(self, obj: t.Any, value: T) -> None: 

63 if self.read_only: 

64 raise AttributeError("read only property") 

65 

66 obj.environ[self.name] = self.dump_func(value) 

67 

68 def __delete__(self, obj: t.Any) -> None: 

69 if self.read_only: 

70 raise AttributeError("read only property") 

71 

72 obj.environ.pop(self.name, None) 

73 

74 def __repr__(self) -> str: 

75 return f"<environ_property {self.name}>" 

76 

77 

78class header_property(t.Generic[T]): 

79 """A property that returns a key from ``headers``. 

80 

81 .. deprecated:: 3.2 

82 Will be removed in Werkzeug 4.0. Access ``headers`` directly instead. 

83 """ 

84 

85 def __init__( 

86 self, 

87 name: str, 

88 default: T | None = None, 

89 load_func: t.Callable[[str], T] | None = None, 

90 dump_func: t.Callable[[T | t.Any], str | None] = str, 

91 read_only: bool = False, 

92 doc: str | None = None, 

93 ) -> None: 

94 self.name = name 

95 self.default = default 

96 self.load_func = load_func 

97 self.dump_func = dump_func 

98 self.read_only = read_only 

99 

100 if doc is not None: 

101 self.__doc__ = doc 

102 

103 @t.overload 

104 def __get__(self, obj: None, cls: type[t.Any], /) -> t.Self: ... 

105 @t.overload 

106 def __get__(self, obj: t.Any, cls: type[t.Any] | None = ..., /) -> T: ... 

107 def __get__( 

108 self, obj: t.Any | None = None, cls: type[t.Any] | None = None, / 

109 ) -> T | t.Self: 

110 if obj is None: 

111 return self 

112 

113 try: 

114 value: t.Any = obj.headers[self.name] 

115 except KeyError: 

116 value = self.default 

117 else: 

118 if self.load_func is not None: 

119 try: 

120 value = self.load_func(value) 

121 except (ValueError, TypeError): 

122 value = self.default 

123 

124 if self.read_only: 

125 # Cache to avoid repeated calls. 

126 obj.__dict__[self.name] = value 

127 

128 return value # type: ignore[no-any-return] 

129 

130 def __set__(self, obj: t.Any, value: T | t.Any | None) -> None: 

131 if self.read_only: 

132 raise AttributeError("read only property") 

133 

134 if value is None: 

135 del obj.headers[self.name] 

136 return 

137 

138 result = self.dump_func(value) 

139 

140 if not result: 

141 del obj.headers[self.name] 

142 return 

143 

144 obj.headers[self.name] = result 

145 

146 def __delete__(self, obj: t.Any) -> None: 

147 if self.read_only: 

148 # Clear the cache. 

149 obj.__dict__.pop(self.name, None) 

150 else: 

151 del obj.headers[self.name] 

152 

153 def __repr__(self) -> str: 

154 return f"<header_property {self.name}>" 

155 

156 

157class _DSProto(t.Protocol): 

158 _on_update: t.Callable[[t.Self], None] | None 

159 

160 @classmethod 

161 def from_header(cls, value: str | None) -> t.Self | None: ... 

162 

163 def to_header(self) -> str: ... 

164 

165 

166_DS = t.TypeVar("_DS", bound=_DSProto) 

167 

168 

169class structure_property(t.Generic[_DS]): 

170 """A header property that uses a structured header class from 

171 ``werkzeug.datastructures``. Uses the ``from_header`` classmethod on get, 

172 and the ``to_header`` method on set. When getting or setting, the instance's 

173 ``_on_update`` callback is set so that modifying it updates the header. 

174 Deleting, setting to ``None`` or empty, or modifying to empty all unset the 

175 header. 

176 

177 ``HeaderSet`` is special cased to automatically support setting to a 

178 collection (``set``, ``list``, ``tuple``). 

179 

180 This is only needed for ``Response``. ``header_property`` is sufficient for 

181 ``Request`` since it doesn't need to set up ``on_update`` or assignment. 

182 

183 :param key: The header key. 

184 :param cls: The structured header class. 

185 :param doc: The documentation about the header. 

186 :param deprecate_str: Whether to show a deprecation warning on setting 

187 ``str``. Otherwise, setting ``str`` is not allowed. 

188 """ 

189 

190 def __init__( 

191 self, key: str, cls: type[_DS], doc: str, deprecate_str: bool = False 

192 ) -> None: 

193 self.key = key 

194 self.cls = cls 

195 self.__doc__ = doc 

196 self.fset: t.Callable[[Response, t.Any], None] | None = None 

197 self.deprecate_str = deprecate_str 

198 

199 if cls is HeaderSet: 

200 self.fset = make_set_header_set(key) 

201 

202 def register_setter(self, fset: t.Callable[[Response, t.Any], None]) -> t.Self: 

203 """Use this setter function instead of the built-in behavior. This will 

204 only be triggered if the value is not ``None`` or false. It must handle 

205 setting ``headers`` and ``_on_update``. 

206 """ 

207 self.fset = fset 

208 return self 

209 

210 def __repr__(self) -> str: 

211 return f"<{self.key} header>" 

212 

213 @t.overload 

214 def __get__(self, obj: None, cls: type[Response], /) -> t.Self: ... 

215 @t.overload 

216 def __get__(self, obj: t.Any, cls: type[Response] | None = ..., /) -> _DS: ... 

217 def __get__( 

218 self, obj: Response | None, owner: type[Response] | None = None, / 

219 ) -> _DS | t.Self: 

220 if obj is None: 

221 return self 

222 

223 if (value := self.cls.from_header(obj.headers.get(self.key))) is None: 

224 return None # type: ignore[return-value] 

225 

226 value._on_update = make_structure_on_update(obj, self.key, self.cls) 

227 return value 

228 

229 def __set__(self, obj: Response, value: _DS | t.Any | None) -> None: 

230 if not value: 

231 del obj.headers[self.key] 

232 elif self.deprecate_str and isinstance(value, str): 

233 import warnings 

234 

235 warnings.warn( 

236 f"Setting '{self.key.lower().replace('-', '_')}' to a string is" 

237 " deprecated and will not be supported in Werkzeug 4.0. Set a" 

238 f" '{self.cls.__name__}' instance, or set the string in 'headers'" 

239 " instead.", 

240 DeprecationWarning, 

241 stacklevel=2, 

242 ) 

243 obj.headers[self.key] = value 

244 elif self.fset is not None: 

245 self.fset(obj, value) 

246 else: 

247 obj.headers[self.key] = value.to_header() 

248 value._on_update = make_structure_on_update(obj, self.key, self.cls) 

249 

250 def __delete__(self, obj: Response) -> None: 

251 del obj.headers[self.key] 

252 

253 

254def make_set_header_set( 

255 key: str, 

256) -> t.Callable[[Response, HeaderSet | t.Collection[str]], None]: 

257 """Allow setting ``HeaderSet`` to a collection (``set``, ``list``, ``tuple``). 

258 

259 :param key: The header key. 

260 """ 

261 

262 def set_header_set( 

263 response: Response, value: HeaderSet | t.Collection[str] 

264 ) -> None: 

265 if isinstance(value, HeaderSet): 

266 response.headers[key] = value.to_header() 

267 else: 

268 response.headers[key] = dump_header(value) 

269 

270 return set_header_set 

271 

272 

273def make_structure_on_update( 

274 response: Response, key: str, cls: type[_DS] 

275) -> t.Callable[[_DS], None]: 

276 """Factory for ``on_update`` functions. 

277 

278 :param response: The response instance to set the header on. 

279 :param key: The header key. 

280 :param cls: The structured header class. Used for type annotation. 

281 """ 

282 

283 def on_update(value: _DS) -> None: 

284 if not value: 

285 del response.headers[key] 

286 else: 

287 response.headers[key] = value.to_header() 

288 

289 return on_update