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()}>"