Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tornado/util.py: 34%
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
1"""Miscellaneous utility functions and classes.
3This module is used internally by Tornado. It is not necessarily expected
4that the functions and classes defined here will be useful to other
5applications, but they are documented here in case they are.
7The one public-facing part of this module is the `Configurable` class
8and its `~Configurable.configure` method, which becomes a part of the
9interface of its subclasses, including `.AsyncHTTPClient`, `.IOLoop`,
10and `.Resolver`.
11"""
13from __future__ import annotations
15import array
16import asyncio
17import os
18import re
19import typing
20import warnings
21import zlib
22from collections.abc import Callable, Mapping, Sequence
23from inspect import getfullargspec
24from re import Match
25from typing import (
26 Any,
27)
29if typing.TYPE_CHECKING:
30 # Additional imports only used in type comments.
31 # This lets us make these imports lazy.
32 import datetime
33 import unittest
34 from types import TracebackType
36# Aliases for types that are spelled differently in different Python
37# versions. bytes_type is deprecated and no longer used in Tornado
38# itself but is left in case anyone outside Tornado is using it.
39bytes_type = bytes
40unicode_type = str
41basestring_type = str
44# versionchanged:: 6.2
45# no longer our own TimeoutError, use standard asyncio class (which is the
46# builtin TimeoutError since Python 3.11). Kept as an alias for compatibility;
47# new code should use the builtin TimeoutError.
48TimeoutError = asyncio.TimeoutError
51class ObjectDict(dict[str, Any]):
52 """Makes a dictionary behave like an object, with attribute-style access."""
54 def __getattr__(self, name: str) -> Any:
55 try:
56 return self[name]
57 except KeyError:
58 raise AttributeError(name)
60 def __setattr__(self, name: str, value: Any) -> None:
61 self[name] = value
64class GzipDecompressor:
65 """Streaming gzip decompressor.
67 The interface is like that of `zlib.decompressobj` (without some of the
68 optional arguments, but it understands gzip headers and checksums.
70 .. versionchanged:: 6.6
72 Streams containing multiple gzip members (as defined in :rfc:`1952`
73 section 2.2) are now decompressed in full; previously everything
74 after the first member was ignored. Any data after the last member
75 that is not the start of another member is now an error, as is a
76 stream that ends partway through a member.
77 """
79 def __init__(self) -> None:
80 # The decompressor for the member currently being read. It is None
81 # before the first member and between members, which are the only
82 # places the stream may validly end. (We can't create one up front
83 # and check its eof attribute instead, because a new decompressobj
84 # is not at eof, and an empty stream, such as the body of a
85 # response to a HEAD request, is valid, as it is for
86 # `gzip.decompress`.)
87 self._decompressobj: zlib._Decompress | None = None
88 # Input that has not been passed to ``_decompressobj``, because
89 # ``max_length`` cut a call short. This is what `unconsumed_tail`
90 # reports; callers must pass it back to `decompress`.
91 self._unconsumed_tail = b""
93 @staticmethod
94 def _new_decompressobj() -> "zlib._Decompress":
95 # Magic parameter makes zlib module understand gzip header
96 # http://stackoverflow.com/questions/1838699/how-can-i-decompress-a-gzip-stream-with-zlib
97 # This works on cpython and pypy, but not jython.
98 return zlib.decompressobj(16 + zlib.MAX_WBITS)
100 def decompress(self, value: bytes, max_length: int = 0) -> bytes:
101 """Decompress a chunk, returning newly-available data.
103 When there is no more input, call `flush` to check that the stream
104 was complete.
106 If ``max_length`` is given, some input data may be left over
107 in ``unconsumed_tail``; you must retrieve this value and pass
108 it back to a future call to `decompress` if it is not empty.
109 A call may return less than ``max_length``, or nothing at all,
110 even though ``unconsumed_tail`` is not empty (for example, at the
111 end of one gzip member and the start of the next), so keep
112 calling `decompress` until ``unconsumed_tail`` is empty.
114 Raises `zlib.error` if the input is not valid gzip data.
116 .. deprecated:: 6.6
118 Calling this method without ``max_length`` (or with
119 ``max_length=0``) is deprecated, because a small input can
120 decompress to an arbitrarily large output. In Tornado 7.0,
121 ``max_length`` will be required.
122 """
123 if not max_length:
124 warnings.warn(
125 "GzipDecompressor.decompress without max_length is deprecated",
126 DeprecationWarning,
127 stacklevel=2,
128 )
129 result = self._decompress(value, max_length)
130 if not max_length:
131 # Without a limit, callers expect all the output at once, so we
132 # can't leave the start of the next member in unconsumed_tail.
133 while self._unconsumed_tail:
134 result += self._decompress(self._unconsumed_tail, 0)
135 return result
137 def _decompress(self, value: bytes, max_length: int) -> bytes:
138 # Each call makes one call to zlib; the caller loops over
139 # unconsumed_tail. This follows `gzip.decompress`, except that the
140 # header, CRC and length are checked by zlib itself.
141 if not value:
142 # Don't start a member (or report it as truncated) without data.
143 self._unconsumed_tail = b""
144 return b""
145 if self._decompressobj is None:
146 # This is the start of a member. zlib would reject anything
147 # else too, but with a less helpful message. value may be a
148 # single byte, in which case zlib checks the second one when it
149 # arrives.
150 if not b"\x1f\x8b".startswith(value[:2]):
151 raise zlib.error("expected a gzip member, got %r" % value[:2])
152 self._decompressobj = self._new_decompressobj()
153 result = self._decompressobj.decompress(value, max_length)
154 if self._decompressobj.eof:
155 # The rest of the input is the next member (or an error), so
156 # the caller must pass it back just like input that max_length
157 # held back. It is in unused_data; unconsumed_tail may hold a
158 # stale copy of it (a CPython bug), so don't look at that.
159 self._unconsumed_tail = self._decompressobj.unused_data
160 self._decompressobj = None
161 else:
162 self._unconsumed_tail = self._decompressobj.unconsumed_tail
163 return result
165 @property
166 def unconsumed_tail(self) -> bytes:
167 """Returns the unconsumed portion left over"""
168 return self._unconsumed_tail
170 def flush(self) -> bytes:
171 """Signals the end of the input.
173 Call this once, after the last call to `decompress`, to check that
174 the stream was complete. Raises `zlib.error` if the input ended
175 partway through a gzip member. Raises `ValueError` if
176 `unconsumed_tail` is not empty; it must be passed back to
177 `decompress` first.
179 Always returns an empty byte string, because `decompress` returns
180 all of the output. The return value is kept for compatibility with
181 `zlib.Decompress.flush`, whose interface this follows.
183 No other methods may be called on this object after `flush`.
185 .. versionchanged:: 6.6
187 Previously, this decompressed any remaining ``unconsumed_tail``
188 with no limit on the size of the output, and it did not detect
189 truncated input.
190 """
191 if self._unconsumed_tail:
192 raise ValueError(
193 "unconsumed_tail must be passed to decompress before flush"
194 )
195 if self._decompressobj is not None:
196 # A member has started but not ended. zlib is not holding back
197 # output that flushing it would return: when max_length cuts it
198 # off, the rest of the member, which includes at least its
199 # 8-byte trailer, is left in unconsumed_tail, which is empty. So
200 # zlib stopped because it ran out of input.
201 raise zlib.error("incomplete or truncated gzip stream")
202 return b""
205def import_object(name: str) -> Any:
206 """Imports an object by name.
208 ``import_object('x')`` is equivalent to ``import x``.
209 ``import_object('x.y.z')`` is equivalent to ``from x.y import z``.
211 >>> import tornado.escape
212 >>> import_object('tornado.escape') is tornado.escape
213 True
214 >>> import_object('tornado.escape.utf8') is tornado.escape.utf8
215 True
216 >>> import_object('tornado') is tornado
217 True
218 >>> import_object('tornado.missing_module')
219 Traceback (most recent call last):
220 ...
221 ImportError: No module named missing_module
222 """
223 if name.count(".") == 0:
224 return __import__(name)
226 parts = name.split(".")
227 obj = __import__(".".join(parts[:-1]), fromlist=[parts[-1]])
228 try:
229 return getattr(obj, parts[-1])
230 except AttributeError:
231 raise ImportError("No module named %s" % parts[-1])
234def exec_in(
235 code: Any, glob: dict[str, Any], loc: Mapping[str, Any] | None | None = None
236) -> None:
237 if isinstance(code, str):
238 # exec(string) inherits the caller's future imports; compile
239 # the string first to prevent that.
240 code = compile(code, "<string>", "exec", dont_inherit=True)
241 exec(code, glob, loc)
244def raise_exc_info(
245 exc_info: tuple[type | None, BaseException | None, TracebackType | None],
246) -> typing.NoReturn:
247 try:
248 if exc_info[1] is not None:
249 raise exc_info[1].with_traceback(exc_info[2])
250 else:
251 raise TypeError("raise_exc_info called with no exception")
252 finally:
253 # Clear the traceback reference from our stack frame to
254 # minimize circular references that slow down GC.
255 exc_info = (None, None, None)
258def errno_from_exception(e: BaseException) -> int | None:
259 """Provides the errno from an Exception object.
261 There are cases that the errno attribute was not set so we pull
262 the errno out of the args but if someone instantiates an Exception
263 without any args you will get a tuple error. So this function
264 abstracts all that behavior to give you a safe way to get the
265 errno.
266 """
268 if hasattr(e, "errno"):
269 return e.errno # type: ignore
270 elif e.args:
271 return e.args[0]
272 else:
273 return None
276_alphanum = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
279def _re_unescape_replacement(match: Match[str]) -> str:
280 group = match.group(1)
281 if group[0] in _alphanum:
282 raise ValueError("cannot unescape '\\\\%s'" % group[0])
283 return group
286_re_unescape_pattern = re.compile(r"\\(.)", re.DOTALL)
289def re_unescape(s: str) -> str:
290 r"""Unescape a string escaped by `re.escape`.
292 May raise ``ValueError`` for regular expressions which could not
293 have been produced by `re.escape` (for example, strings containing
294 ``\d`` cannot be unescaped).
296 .. versionadded:: 4.4
297 """
298 return _re_unescape_pattern.sub(_re_unescape_replacement, s)
301class Configurable:
302 """Base class for configurable interfaces.
304 A configurable interface is an (abstract) class whose constructor
305 acts as a factory function for one of its implementation subclasses.
306 The implementation subclass as well as optional keyword arguments to
307 its initializer can be set globally at runtime with `configure`.
309 By using the constructor as the factory method, the interface
310 looks like a normal class, `isinstance` works as usual, etc. This
311 pattern is most useful when the choice of implementation is likely
312 to be a global decision (e.g. when `~select.epoll` is available,
313 always use it instead of `~select.select`), or when a
314 previously-monolithic class has been split into specialized
315 subclasses.
317 Configurable subclasses must define the class methods
318 `configurable_base` and `configurable_default`, and use the instance
319 method `initialize` instead of ``__init__``.
321 .. versionchanged:: 5.0
323 It is now possible for configuration to be specified at
324 multiple levels of a class hierarchy.
326 """
328 # Type annotations on this class are mostly done with comments
329 # because they need to refer to Configurable, which isn't defined
330 # until after the class definition block. These can use regular
331 # annotations when our minimum python version is 3.7.
332 #
333 # There may be a clever way to use generics here to get more
334 # precise types (i.e. for a particular Configurable subclass T,
335 # all the types are subclasses of T, not just Configurable).
336 __impl_class: type[Configurable] | None = None
337 __impl_kwargs: dict[str, Any] | None = None
339 def __new__(cls, *args: Any, **kwargs: Any) -> Any:
340 base = cls.configurable_base()
341 init_kwargs: dict[str, Any] = {}
342 if cls is base:
343 impl = cls.configured_class()
344 if base.__impl_kwargs:
345 init_kwargs.update(base.__impl_kwargs)
346 else:
347 impl = cls
348 init_kwargs.update(kwargs)
349 if impl.configurable_base() is not base:
350 # The impl class is itself configurable, so recurse.
351 return impl(*args, **init_kwargs)
352 instance = super().__new__(impl)
353 # initialize vs __init__ chosen for compatibility with AsyncHTTPClient
354 # singleton magic. If we get rid of that we can switch to __init__
355 # here too.
356 instance.initialize(*args, **init_kwargs)
357 return instance
359 @classmethod
360 def configurable_base(cls) -> type[Configurable]:
361 """Returns the base class of a configurable hierarchy.
363 This will normally return the class in which it is defined.
364 (which is *not* necessarily the same as the ``cls`` classmethod
365 parameter).
367 """
368 raise NotImplementedError()
370 @classmethod
371 def configurable_default(cls) -> type[Configurable]:
372 """Returns the implementation class to be used if none is configured."""
373 raise NotImplementedError()
375 def _initialize(self) -> None:
376 pass
378 initialize: Callable[..., None] = _initialize
379 """Initialize a `Configurable` subclass instance.
381 Configurable classes should use `initialize` instead of ``__init__``.
383 .. versionchanged:: 4.2
384 Now accepts positional arguments in addition to keyword arguments.
385 """
387 @classmethod
388 def configure(cls, impl: None | str | type[Configurable], **kwargs: Any) -> None:
389 """Sets the class to use when the base class is instantiated.
391 Keyword arguments will be saved and added to the arguments passed
392 to the constructor. This can be used to set global defaults for
393 some parameters.
394 """
395 base = cls.configurable_base()
396 if isinstance(impl, str):
397 impl = typing.cast(type[Configurable], import_object(impl))
398 if impl is not None and not issubclass(impl, cls):
399 raise ValueError("Invalid subclass of %s" % cls)
400 base.__impl_class = impl
401 base.__impl_kwargs = kwargs
403 @classmethod
404 def configured_class(cls) -> type[Configurable]:
405 """Returns the currently configured class."""
406 base = cls.configurable_base()
407 # Manually mangle the private name to see whether this base
408 # has been configured (and not another base higher in the
409 # hierarchy).
410 if base.__dict__.get("_Configurable__impl_class") is None:
411 base.__impl_class = cls.configurable_default()
412 if base.__impl_class is not None:
413 return base.__impl_class
414 else:
415 # Should be impossible, but mypy wants an explicit check.
416 raise ValueError("configured class not found")
418 @classmethod
419 def _save_configuration(
420 cls,
421 ) -> tuple[type[Configurable] | None, dict[str, Any] | None]:
422 base = cls.configurable_base()
423 return (base.__impl_class, base.__impl_kwargs)
425 @classmethod
426 def _restore_configuration(
427 cls, saved: tuple[type[Configurable] | None, dict[str, Any] | None]
428 ) -> None:
429 base = cls.configurable_base()
430 base.__impl_class = saved[0]
431 base.__impl_kwargs = saved[1]
434class ArgReplacer:
435 """Replaces one value in an ``args, kwargs`` pair.
437 Inspects the function signature to find an argument by name
438 whether it is passed by position or keyword. For use in decorators
439 and similar wrappers.
440 """
442 def __init__(self, func: Callable, name: str) -> None:
443 self.name = name
444 try:
445 self.arg_pos: int | None = self._getargnames(func).index(name)
446 except ValueError:
447 # Not a positional parameter
448 self.arg_pos = None
450 def _getargnames(self, func: Callable) -> list[str]:
451 try:
452 return getfullargspec(func).args
453 except TypeError:
454 if hasattr(func, "func_code"):
455 # Cython-generated code has all the attributes needed
456 # by inspect.getfullargspec, but the inspect module only
457 # works with ordinary functions. Inline the portion of
458 # getfullargspec that we need here. Note that for static
459 # functions the @cython.binding(True) decorator must
460 # be used (for methods it works out of the box).
461 code = func.func_code # type: ignore
462 return code.co_varnames[: code.co_argcount]
463 raise
465 def get_old_value(
466 self, args: Sequence[Any], kwargs: dict[str, Any], default: Any = None
467 ) -> Any:
468 """Returns the old value of the named argument without replacing it.
470 Returns ``default`` if the argument is not present.
471 """
472 if self.arg_pos is not None and len(args) > self.arg_pos:
473 return args[self.arg_pos]
474 else:
475 return kwargs.get(self.name, default)
477 def replace(
478 self, new_value: Any, args: Sequence[Any], kwargs: dict[str, Any]
479 ) -> tuple[Any, Sequence[Any], dict[str, Any]]:
480 """Replace the named argument in ``args, kwargs`` with ``new_value``.
482 Returns ``(old_value, args, kwargs)``. The returned ``args`` and
483 ``kwargs`` objects may not be the same as the input objects, or
484 the input objects may be mutated.
486 If the named argument was not found, ``new_value`` will be added
487 to ``kwargs`` and None will be returned as ``old_value``.
488 """
489 if self.arg_pos is not None and len(args) > self.arg_pos:
490 # The arg to replace is passed positionally
491 old_value = args[self.arg_pos]
492 args = list(args) # *args is normally a tuple
493 args[self.arg_pos] = new_value
494 else:
495 # The arg to replace is either omitted or passed by keyword.
496 old_value = kwargs.get(self.name)
497 kwargs[self.name] = new_value
498 return old_value, args, kwargs
501def timedelta_to_seconds(td: datetime.timedelta) -> float:
502 """Equivalent to ``td.total_seconds()`` (introduced in Python 2.7)."""
503 return td.total_seconds()
506def _websocket_mask_python(mask: bytes, data: bytes) -> bytes:
507 """Websocket masking function.
509 `mask` is a `bytes` object of length 4; `data` is a `bytes` object of any length.
510 Returns a `bytes` object of the same length as `data` with the mask applied
511 as specified in section 5.3 of RFC 6455.
513 This pure-python implementation may be replaced by an optimized version when available.
514 """
515 if len(mask) != 4:
516 raise ValueError("mask must be 4 bytes")
517 mask_arr = array.array("B", mask)
518 unmasked_arr = array.array("B", data)
519 for i in range(len(data)):
520 unmasked_arr[i] = unmasked_arr[i] ^ mask_arr[i % 4]
521 return unmasked_arr.tobytes()
524if os.environ.get("TORNADO_NO_EXTENSION") or os.environ.get("TORNADO_EXTENSION") == "0":
525 # These environment variables exist to make it easier to do performance
526 # comparisons; they are not guaranteed to remain supported in the future.
527 _websocket_mask = _websocket_mask_python
528else:
529 try:
530 from tornado.speedups import websocket_mask as _websocket_mask
531 except ImportError:
532 if os.environ.get("TORNADO_EXTENSION") == "1":
533 raise
534 _websocket_mask = _websocket_mask_python
537def doctests() -> unittest.TestSuite:
538 import doctest
540 return doctest.DocTestSuite()