Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/datastructures/range.py: 69%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

214 statements  

1from __future__ import annotations 

2 

3import collections.abc as cabc 

4import typing as t 

5from datetime import datetime 

6 

7from .._internal import _plain_int 

8from ..http import http_date 

9from ..http import parse_date 

10from .etag import ETag 

11 

12T = t.TypeVar("T") 

13 

14 

15class IfRange: 

16 """A parsed ``If-Range`` header. Either a strong ETag or a datetime, but not 

17 both. Weak ETag values must not be used. 

18 

19 :attr:`.Request.if_range` returns an instance. 

20 

21 :param etag: An unquoted strong ETag value. 

22 :param date: A timezone-aware datetime. 

23 

24 .. versionadded:: 0.7 

25 """ 

26 

27 def __init__(self, etag: str | None = None, date: datetime | None = None): 

28 self.etag = etag 

29 """An unquoted strong ETag value.""" 

30 

31 self.date = date 

32 """A timezone-aware datetime.""" 

33 

34 def check(self, etag: ETag | None, last_modified: datetime | None) -> bool: 

35 """Check if the condition is met by the response's values. 

36 

37 If :attr:`etag` is set, the strong comparison function is used with 

38 :attr:`.Response.etag`. If :attr:`datetime` is set, it must equal 

39 :attr:`.Response.last_modified`. 

40 

41 :param etag: The response's ETag. Only a strong ETag can match. 

42 :param last_modified: The response's modification time. 

43 

44 .. versionadded:: 3.2 

45 """ 

46 if self.etag is not None: 

47 return etag is not None and not etag.weak and etag.value == self.etag 

48 

49 if self.date is not None: 

50 return last_modified is not None and last_modified == self.date 

51 

52 return False 

53 

54 @classmethod 

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

56 """Parse an ``If-Range`` header value and create an instance of this 

57 class. A weak ETag value is discarded. 

58 

59 .. versionadded:: 3.2 

60 """ 

61 if not value: 

62 return cls() 

63 

64 if (date := parse_date(value)) is not None: 

65 return cls(date=date) 

66 

67 if (etag := ETag.from_header(value)) is None or etag.weak: 

68 return cls() 

69 

70 return cls(etag=etag.value) 

71 

72 def to_header(self) -> str: 

73 """Convert to an ``If-Range`` header value.""" 

74 if self.date is not None: 

75 return http_date(self.date) 

76 

77 if self.etag is not None: 

78 return ETag(self.etag).to_header() 

79 

80 return "" 

81 

82 def __bool__(self) -> bool: 

83 return self.etag is not None or self.date is not None 

84 

85 def __str__(self) -> str: 

86 return self.to_header() 

87 

88 def __repr__(self) -> str: 

89 return f"<{type(self).__name__} {str(self)!r}>" 

90 

91 

92class Range: 

93 """A parsed ``Range`` header. 

94 

95 :attr:`.Request.range` returns an instance, or ``None`` if the header is not 

96 set. 

97 

98 .. versionchanged:: 3.2 

99 Validation is done in ``from_header``. Values passed to the constructor 

100 are assumed valid. 

101 

102 .. versionchanged:: 0.15 

103 Values passed to the constructor are validated. 

104 

105 .. versionadded:: 0.7 

106 """ 

107 

108 def __init__( 

109 self, units: str, ranges: cabc.Sequence[tuple[int, int | None]] 

110 ) -> None: 

111 self.units = units 

112 """The unit being counted. Only ``"bytes"`` is defined.""" 

113 

114 self.ranges = ranges 

115 """The parsed ``(start, stop)`` ranges. ``stop`` is exclusive, unlike 

116 the raw header value. If ``stop`` is ``None``, ``start`` can be positive 

117 to get the remaining units from that offset, or negative to get that 

118 number of units from the end. 

119 """ 

120 

121 def range_for_length(self, length: int | None) -> tuple[int, int] | None: 

122 """Return the ``(start, stop)`` values to use for a 

123 :class:`.ContentRange` header with the given complete length. Or ``None`` 

124 if a valid range cannot be constructed. 

125 

126 This will generate a range regardless of the value of :attr:`units`. 

127 However, :meth:`.Response.make_conditional` can only handle ``bytes``. 

128 

129 If there are multiple ranges in the header, this will only return the 

130 first range. 

131 

132 :param length: The complete length of the content. ``None`` means the 

133 value is unknown, and will return ``None``. 

134 :return: ``(start, stop)`` if the range and length are valid, ``None`` 

135 otherwise. 

136 

137 .. deprecated:: 3.2 

138 Will be removed in Werkzeug 4.0. Use ``make_content_range`` instead. 

139 

140 .. versionchanged:: 3.2 

141 Allows units other than ``bytes``. 

142 

143 Will return the first range if there are multiple. 

144 

145 A suffix range larger than the length starts at 0 instead of invalid. 

146 """ 

147 import warnings 

148 

149 warnings.warn( 

150 "'range_for_length' is deprecated and will be removed in Werkzeug 4.0." 

151 " Use 'make_content_range` instead.", 

152 DeprecationWarning, 

153 stacklevel=2, 

154 ) 

155 return self._private_range_for_length(length) 

156 

157 def _private_range_for_length(self, length: int | None) -> tuple[int, int] | None: 

158 if length is None or not self.ranges: 

159 return None 

160 

161 start, stop = self.ranges[0] 

162 

163 if start < 0: 

164 start = max(0, start + length) 

165 elif start >= length: 

166 return None 

167 

168 if stop is None or stop > length: 

169 stop = length 

170 

171 return start, stop 

172 

173 def make_content_range(self, length: int | None) -> ContentRange | None: 

174 """Create a :class:`.ContentRange` with the given complete length. Or 

175 ``None`` if a valid range cannot be constructed. 

176 

177 This will generate a range regardless of the value of :attr:`units`. 

178 However, :meth:`.Response.make_conditional` can only handle ``bytes``. 

179 

180 If there are multiple ranges in the header, this will only return the 

181 first range. 

182 

183 :param length: The complete length of the content. ``None`` means the 

184 value is unknown, and will return ``None``. 

185 

186 .. versionchanged:: 3.2 

187 Allows units other than ``bytes``. 

188 

189 Will return the first range if there are multiple. 

190 

191 A suffix range larger than the length starts at 0 instead of invalid. 

192 """ 

193 # TODO inline after deprecation 

194 if (bounds := self._private_range_for_length(length)) is None: 

195 return None 

196 

197 return ContentRange(self.units, bounds[0], bounds[1], length) 

198 

199 @classmethod 

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

201 """Parse a ``Range`` header value and create an instance of this class, 

202 or ``None`` if the value is empty. 

203 

204 .. versionadded:: 3.2 

205 """ 

206 if not value or "=" not in value: 

207 return None 

208 

209 ranges = [] 

210 last_end = 0 

211 units, _, ranges_str = value.partition("=") 

212 units = units.strip(" \t").lower() 

213 

214 for item in ranges_str.split(","): 

215 item = item.strip(" \t") 

216 

217 if "-" not in item: 

218 return None 

219 

220 if item.startswith("-"): 

221 if last_end < 0: 

222 return None 

223 

224 try: 

225 begin = _plain_int(item) 

226 except ValueError: 

227 return None 

228 

229 # -0 will parse to 0, and is an invalid suffix length. 

230 if begin == 0: 

231 return None 

232 

233 end = None 

234 last_end = -1 

235 

236 else: 

237 begin_str, _, end_str = item.partition("-") 

238 begin_str = begin_str.strip(" \t") 

239 end_str = end_str.strip(" \t") 

240 

241 try: 

242 begin = _plain_int(begin_str) 

243 except ValueError: 

244 return None 

245 

246 if begin < last_end or last_end < 0: 

247 return None 

248 

249 if end_str: 

250 try: 

251 end = _plain_int(end_str) + 1 

252 except ValueError: 

253 return None 

254 

255 if begin >= end: 

256 return None 

257 else: 

258 end = None 

259 

260 last_end = end if end is not None else -1 

261 

262 ranges.append((begin, end)) 

263 

264 return cls(units, ranges) 

265 

266 def to_header(self) -> str: 

267 """Convert to a ``Range`` header value.""" 

268 ranges = [] 

269 for begin, end in self.ranges: 

270 if end is None: 

271 ranges.append(f"{begin}-" if begin >= 0 else str(begin)) 

272 else: 

273 ranges.append(f"{begin}-{end - 1}") 

274 return f"{self.units}={','.join(ranges)}" 

275 

276 def to_content_range_header(self, length: int | None) -> str | None: 

277 """Converts the object into `Content-Range` HTTP header, 

278 based on given length 

279 

280 .. deprecated:: 3.2 

281 Will be removed in Werkzeug 4.0. Use ``make_content_range`` then 

282 call its ``to_header`` method instead. 

283 

284 .. versionchanged:: 3.2 

285 Allows units other than ``bytes``. 

286 

287 Will return the first range if there are multiple. 

288 

289 A suffix range larger than the length starts at 0 instead of invalid. 

290 """ 

291 import warnings 

292 

293 warnings.warn( 

294 "'to_content_range_header' is deprecated and will be removed in" 

295 " Werkzeug 4.0. Use 'make_content_range` then call its 'to_header'" 

296 " method instead.", 

297 DeprecationWarning, 

298 stacklevel=2, 

299 ) 

300 

301 if (cr := self.make_content_range(length)) is None: 

302 return None 

303 

304 return cr.to_header() 

305 

306 def __str__(self) -> str: 

307 return self.to_header() 

308 

309 def __repr__(self) -> str: 

310 return f"<{type(self).__name__} {str(self)!r}>" 

311 

312 

313class _CallbackProperty(t.Generic[T]): 

314 def __set_name__(self, owner: type[ContentRange], name: str) -> None: 

315 self.attr = f"_{name}" 

316 

317 @t.overload 

318 def __get__(self, instance: None, owner: None) -> t.Self: ... 

319 @t.overload 

320 def __get__(self, instance: ContentRange, owner: type[ContentRange]) -> T: ... 

321 def __get__( 

322 self, instance: ContentRange | None, owner: type[ContentRange] | None 

323 ) -> t.Self | T: 

324 if instance is None: 

325 return self 

326 

327 return instance.__dict__[self.attr] # type: ignore[no-any-return] 

328 

329 def __set__(self, instance: ContentRange, value: T) -> None: 

330 instance.__dict__[self.attr] = value 

331 

332 if instance._on_update is not None: 

333 instance._on_update(instance) 

334 

335 

336class ContentRange: 

337 """A parsed ``Content-Range`` header. 

338 

339 Set :attr:`.Response.content_range` to an instance to set the header. 

340 Modifying the instance will update the header. 

341 

342 .. versionchanged:: 3.2 

343 The ``on_update`` parameter was removed. Argument defaults were added. 

344 Considered false if neither a range nor length is set. 

345 

346 .. versionadded:: 0.7 

347 """ 

348 

349 def __init__( 

350 self, 

351 units: str = "bytes", 

352 start: int | None = None, 

353 stop: int | None = None, 

354 length: int | None = None, 

355 ) -> None: 

356 self._units = units 

357 self._start = start 

358 self._stop = stop 

359 self._length = length 

360 self._on_update: cabc.Callable[[ContentRange], None] | None = None 

361 

362 units: str = _CallbackProperty() # type: ignore[assignment] 

363 """The unit being counted. Only ``"bytes"`` is defined.""" 

364 

365 start: int | None = _CallbackProperty() # type: ignore[assignment] 

366 """The start point, inclusive. ``None`` means the range is unsatisfiable.""" 

367 

368 stop: int | None = _CallbackProperty() # type: ignore[assignment] 

369 """The stop point. Exclusive, unlike the raw header value. ``None`` means 

370 the range is unsatisfiable. 

371 """ 

372 

373 length: int | None = _CallbackProperty() # type: ignore[assignment] 

374 """The complete length of the content. ``None`` means the length is unknown.""" 

375 

376 def set( 

377 self, 

378 start: int | None, 

379 stop: int | None, 

380 length: int | None = None, 

381 units: str = "bytes", 

382 ) -> None: 

383 """Update the header. 

384 

385 .. deprecated:: 3.2 

386 Will be removed in Werkzeug 4.0. Use 

387 ``request.content_range = ContentRange(...)`` instead. 

388 """ 

389 import warnings 

390 

391 warnings.warn( 

392 "The 'set' method is deprecated and will be removed in Werkzeug 4.0." 

393 " Use 'request.content_range = ContentRange(...)' instead.", 

394 DeprecationWarning, 

395 stacklevel=2, 

396 ) 

397 self._units = units 

398 self._start = start 

399 self._stop = stop 

400 self._length = length 

401 

402 if self._on_update is not None: 

403 self._on_update(self) 

404 

405 def unset(self) -> None: 

406 """Unset the header. 

407 

408 .. deprecated:: 3.2 

409 Will be removed in Werkzeug 4.0. Use ``del request.content_range`` 

410 instead. 

411 """ 

412 import warnings 

413 

414 warnings.warn( 

415 "The 'unset' method is deprecated and will be removed in Werkzeug 4.0." 

416 " Use 'del request.content_range' instead.", 

417 DeprecationWarning, 

418 stacklevel=2, 

419 ) 

420 self.set(None, None, units="") 

421 

422 @classmethod 

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

424 """Parse a ``Content-Range`` header value and create an instance of this class. 

425 

426 .. versionadded:: 3.2 

427 """ 

428 if not value: 

429 return cls() 

430 

431 units, _, range_str = value.partition(" ") 

432 range_str, sep, length_str = range_str.partition("/") 

433 

434 if not sep: 

435 return cls() 

436 

437 if length_str == "*": 

438 length = None 

439 else: 

440 try: 

441 length = _plain_int(length_str) 

442 except ValueError: 

443 return cls() 

444 

445 if length < 0: 

446 return cls() 

447 

448 if range_str == "*": 

449 if length_str == "*": 

450 return cls() 

451 

452 return cls(units, None, None, length) 

453 

454 start_str, sep, stop_str = range_str.partition("-") 

455 

456 if not sep: 

457 return cls() 

458 

459 try: 

460 start = _plain_int(start_str) 

461 stop = _plain_int(stop_str) 

462 except ValueError: 

463 return cls() 

464 

465 if not (0 <= start <= stop) or (length is not None and length <= stop): 

466 return cls() 

467 

468 return cls(units, start, stop + 1, length) 

469 

470 def to_header(self) -> str: 

471 """Convert to a ``Content-Range`` header value.""" 

472 if not self: 

473 return "" 

474 

475 length = "*" if self._length is None else self._length 

476 

477 if self._start is None or self._stop is None: 

478 return f"{self._units} */{length}" 

479 

480 return f"{self._units} {self._start}-{self._stop - 1}/{length}" 

481 

482 def __bool__(self) -> bool: 

483 return not ( 

484 (self._start is None or self._stop is None) and self._length is None 

485 ) 

486 

487 def __str__(self) -> str: 

488 return self.to_header() 

489 

490 def __repr__(self) -> str: 

491 return f"<{type(self).__name__} {self.to_header()}>"