codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import collections.abc as cabc4import typing as t5from datetime import datetime6 7if t.TYPE_CHECKING:8 import typing_extensions as te9 10T = t.TypeVar("T")11 12 13class IfRange:14 """Very simple object that represents the `If-Range` header in parsed15 form. It will either have neither a etag or date or one of either but16 never both.17 18 .. versionadded:: 0.719 """20 21 def __init__(self, etag: str | None = None, date: datetime | None = None):22 #: The etag parsed and unquoted. Ranges always operate on strong23 #: etags so the weakness information is not necessary.24 self.etag = etag25 #: The date in parsed format or `None`.26 self.date = date27 28 def to_header(self) -> str:29 """Converts the object back into an HTTP header."""30 if self.date is not None:31 return http.http_date(self.date)32 if self.etag is not None:33 return http.quote_etag(self.etag)34 return ""35 36 def __str__(self) -> str:37 return self.to_header()38 39 def __repr__(self) -> str:40 return f"<{type(self).__name__} {str(self)!r}>"41 42 43class Range:44 """Represents a ``Range`` header. All methods only support only45 bytes as the unit. Stores a list of ranges if given, but the methods46 only work if only one range is provided.47 48 :raise ValueError: If the ranges provided are invalid.49 50 .. versionchanged:: 0.1551 The ranges passed in are validated.52 53 .. versionadded:: 0.754 """55 56 def __init__(57 self, units: str, ranges: cabc.Sequence[tuple[int, int | None]]58 ) -> None:59 #: The units of this range. Usually "bytes".60 self.units = units61 #: A list of ``(begin, end)`` tuples for the range header provided.62 #: The ranges are non-inclusive.63 self.ranges = ranges64 65 for start, end in ranges:66 if start is None or (end is not None and (start < 0 or start >= end)):67 raise ValueError(f"{(start, end)} is not a valid range.")68 69 def range_for_length(self, length: int | None) -> tuple[int, int] | None:70 """If the range is for bytes, the length is not None and there is71 exactly one range and it is satisfiable it returns a ``(start, stop)``72 tuple, otherwise `None`.73 """74 if self.units != "bytes" or length is None or len(self.ranges) != 1:75 return None76 start, end = self.ranges[0]77 if end is None:78 end = length79 if start < 0:80 start += length81 if http.is_byte_range_valid(start, end, length):82 return start, min(end, length)83 return None84 85 def make_content_range(self, length: int | None) -> ContentRange | None:86 """Creates a :class:`~werkzeug.datastructures.ContentRange` object87 from the current range and given content length.88 """89 rng = self.range_for_length(length)90 if rng is not None:91 return ContentRange(self.units, rng[0], rng[1], length)92 return None93 94 def to_header(self) -> str:95 """Converts the object back into an HTTP header."""96 ranges = []97 for begin, end in self.ranges:98 if end is None:99 ranges.append(f"{begin}-" if begin >= 0 else str(begin))100 else:101 ranges.append(f"{begin}-{end - 1}")102 return f"{self.units}={','.join(ranges)}"103 104 def to_content_range_header(self, length: int | None) -> str | None:105 """Converts the object into `Content-Range` HTTP header,106 based on given length107 """108 range = self.range_for_length(length)109 if range is not None:110 return f"{self.units} {range[0]}-{range[1] - 1}/{length}"111 return None112 113 def __str__(self) -> str:114 return self.to_header()115 116 def __repr__(self) -> str:117 return f"<{type(self).__name__} {str(self)!r}>"118 119 120class _CallbackProperty(t.Generic[T]):121 def __set_name__(self, owner: type[ContentRange], name: str) -> None:122 self.attr = f"_{name}"123 124 @t.overload125 def __get__(self, instance: None, owner: None) -> te.Self: ...126 @t.overload127 def __get__(self, instance: ContentRange, owner: type[ContentRange]) -> T: ...128 def __get__(129 self, instance: ContentRange | None, owner: type[ContentRange] | None130 ) -> te.Self | T:131 if instance is None:132 return self133 134 return instance.__dict__[self.attr] # type: ignore[no-any-return]135 136 def __set__(self, instance: ContentRange, value: T) -> None:137 instance.__dict__[self.attr] = value138 139 if instance.on_update is not None:140 instance.on_update(instance)141 142 143class ContentRange:144 """Represents the content range header.145 146 .. versionadded:: 0.7147 """148 149 def __init__(150 self,151 units: str | None,152 start: int | None,153 stop: int | None,154 length: int | None = None,155 on_update: cabc.Callable[[ContentRange], None] | None = None,156 ) -> None:157 self.on_update = on_update158 self.set(start, stop, length, units)159 160 #: The units to use, usually "bytes"161 units: str | None = _CallbackProperty() # type: ignore[assignment]162 #: The start point of the range or `None`.163 start: int | None = _CallbackProperty() # type: ignore[assignment]164 #: The stop point of the range (non-inclusive) or `None`. Can only be165 #: `None` if also start is `None`.166 stop: int | None = _CallbackProperty() # type: ignore[assignment]167 #: The length of the range or `None`.168 length: int | None = _CallbackProperty() # type: ignore[assignment]169 170 def set(171 self,172 start: int | None,173 stop: int | None,174 length: int | None = None,175 units: str | None = "bytes",176 ) -> None:177 """Simple method to update the ranges."""178 assert http.is_byte_range_valid(start, stop, length), "Bad range provided"179 self._units: str | None = units180 self._start: int | None = start181 self._stop: int | None = stop182 self._length: int | None = length183 if self.on_update is not None:184 self.on_update(self)185 186 def unset(self) -> None:187 """Sets the units to `None` which indicates that the header should188 no longer be used.189 """190 self.set(None, None, units=None)191 192 def to_header(self) -> str:193 if self._units is None:194 return ""195 if self._length is None:196 length: str | int = "*"197 else:198 length = self._length199 if self._start is None:200 return f"{self._units} */{length}"201 return f"{self._units} {self._start}-{self._stop - 1}/{length}" # type: ignore[operator]202 203 def __bool__(self) -> bool:204 return self._units is not None205 206 def __str__(self) -> str:207 return self.to_header()208 209 def __repr__(self) -> str:210 return f"<{type(self).__name__} {str(self)!r}>"211 212 213# circular dependencies214from .. import http # noqa: E402215 