basestring_type = basestring # noqa
-class ObjectDict(dict):
+try:
+ import typing # noqa
+ from typing import cast
+
+ _ObjectDictBase = typing.Dict[str, typing.Any]
+except ImportError:
+ _ObjectDictBase = dict
+
+ def cast(typ, x):
+ return x
+else:
+ # More imports that are only needed in type comments.
+ import datetime # noqa
+ import types # noqa
+ from typing import Any, AnyStr, Union, Optional, Dict, Mapping # noqa
+ from typing import Tuple, Match, Callable # noqa
+
+ if PY3:
+ _BaseString = str
+ else:
+ _BaseString = Union[bytes, unicode_type]
+
+
+class ObjectDict(_ObjectDictBase):
"""Makes a dictionary behave like an object, with attribute-style access.
"""
def __getattr__(self, name):
+ # type: (str) -> Any
try:
return self[name]
except KeyError:
raise AttributeError(name)
def __setattr__(self, name, value):
+ # type: (str, Any) -> None
self[name] = value
self.decompressobj = zlib.decompressobj(16 + zlib.MAX_WBITS)
def decompress(self, value, max_length=None):
+ # type: (bytes, Optional[int]) -> bytes
"""Decompress a chunk, returning newly-available data.
Some data may be buffered for later processing; `flush` must
@property
def unconsumed_tail(self):
+ # type: () -> bytes
"""Returns the unconsumed portion left over
"""
return self.decompressobj.unconsumed_tail
def flush(self):
+ # type: () -> bytes
"""Return any remaining buffered data not yet returned by decompress.
Also checks for errors such as truncated input.
def import_object(name):
+ # type: (_BaseString) -> Any
"""Imports an object by name.
import_object('x') is equivalent to 'import x'.
...
ImportError: No module named missing_module
"""
- if isinstance(name, unicode_type) and str is not unicode_type:
- # On python 2 a byte string is required.
+ if not isinstance(name, str):
+ # on python 2 a byte string is required.
name = name.encode('utf-8')
if name.count('.') == 0:
return __import__(name, None, None)
# Stubs to make mypy happy (and later for actual type-checking).
def raise_exc_info(exc_info):
+ # type: (Tuple[type, BaseException, types.TracebackType]) -> None
pass
def exec_in(code, glob, loc=None):
+ # type: (Any, Dict[str, Any], Optional[Mapping[str, Any]]) -> Any
pass
def errno_from_exception(e):
+ # type: (BaseException) -> Optional[int]
"""Provides the errno from an Exception object.
There are cases that the errno attribute was not set so we pull
"""
if hasattr(e, 'errno'):
- return e.errno
+ return e.errno # type: ignore
elif e.args:
return e.args[0]
else:
def _re_unescape_replacement(match):
+ # type: (Match[str]) -> str
group = match.group(1)
if group[0] in _alphanum:
raise ValueError("cannot unescape '\\\\%s'" % group[0])
def re_unescape(s):
+ # type: (str) -> str
"""Unescape a string escaped by `re.escape`.
May raise ``ValueError`` for regular expressions which could not
method `initialize` instead of ``__init__``.
"""
__impl_class = None # type: type
- __impl_kwargs = None # type: dict
+ __impl_kwargs = None # type: Dict[str, Any]
def __new__(cls, *args, **kwargs):
base = cls.configurable_base()
@classmethod
def configurable_base(cls):
+ # type: () -> Any
+ # TODO: This class needs https://github.com/python/typing/issues/107
+ # to be fully typeable.
"""Returns the base class of a configurable hierarchy.
This will normally return the class in which it is defined.
@classmethod
def configurable_default(cls):
+ # type: () -> type
"""Returns the implementation class to be used if none is configured."""
raise NotImplementedError()
def initialize(self):
+ # type: () -> None
"""Initialize a `Configurable` subclass instance.
Configurable classes should use `initialize` instead of ``__init__``.
@classmethod
def configure(cls, impl, **kwargs):
+ # type: (Any, **Any) -> None
"""Sets the class to use when the base class is instantiated.
Keyword arguments will be saved and added to the arguments passed
some parameters.
"""
base = cls.configurable_base()
- if isinstance(impl, (unicode_type, bytes)):
+ if isinstance(impl, (str, unicode_type)):
impl = import_object(impl)
if impl is not None and not issubclass(impl, cls):
raise ValueError("Invalid subclass of %s" % cls)
@classmethod
def configured_class(cls):
+ # type: () -> type
"""Returns the currently configured class."""
base = cls.configurable_base()
if cls.__impl_class is None:
@classmethod
def _save_configuration(cls):
+ # type: () -> Tuple[type, Dict[str, Any]]
base = cls.configurable_base()
return (base.__impl_class, base.__impl_kwargs)
@classmethod
def _restore_configuration(cls, saved):
+ # type: (Tuple[type, Dict[str, Any]]) -> None
base = cls.configurable_base()
base.__impl_class = saved[0]
base.__impl_kwargs = saved[1]
and similar wrappers.
"""
def __init__(self, func, name):
+ # type: (Callable, str) -> None
self.name = name
try:
self.arg_pos = self._getargnames(func).index(name)
self.arg_pos = None
def _getargnames(self, func):
+ # type: (Callable) -> List[str]
try:
return getargspec(func).args
except TypeError:
# getargspec that we need here. Note that for static
# functions the @cython.binding(True) decorator must
# be used (for methods it works out of the box).
- code = func.func_code
+ code = func.func_code # type: ignore
return code.co_varnames[:code.co_argcount]
raise
def get_old_value(self, args, kwargs, default=None):
+ # type: (List[Any], Dict[str, Any], Any) -> Any
"""Returns the old value of the named argument without replacing it.
Returns ``default`` if the argument is not present.
return kwargs.get(self.name, default)
def replace(self, new_value, args, kwargs):
+ # type: (Any, List[Any], Dict[str, Any]) -> Tuple[Any, List[Any], Dict[str, Any]]
"""Replace the named argument in ``args, kwargs`` with ``new_value``.
Returns ``(old_value, args, kwargs)``. The returned ``args`` and
def timedelta_to_seconds(td):
+ # type: (datetime.timedelta) -> float
"""Equivalent to td.total_seconds() (introduced in python 2.7)."""
return (td.microseconds + (td.seconds + td.days * 24 * 3600) * 10 ** 6) / float(10 ** 6)
def _websocket_mask_python(mask, data):
+ # type: (bytes, bytes) -> bytes
"""Websocket masking function.
`mask` is a `bytes` object of length 4; `data` is a `bytes` object of any length.
This pure-python implementation may be replaced by an optimized version when available.
"""
- mask = array.array("B", mask)
- unmasked = array.array("B", data)
+ mask_arr = array.array("B", mask)
+ unmasked_arr = array.array("B", data)
for i in xrange(len(data)):
- unmasked[i] = unmasked[i] ^ mask[i % 4]
- if hasattr(unmasked, 'tobytes'):
+ unmasked_arr[i] = unmasked_arr[i] ^ mask_arr[i % 4]
+ if PY3:
# tostring was deprecated in py32. It hasn't been removed,
# but since we turn on deprecation warnings in our tests
# we need to use the right one.
- return unmasked.tobytes()
+ return unmasked_arr.tobytes()
else:
- return unmasked.tostring()
+ return unmasked_arr.tostring()
if (os.environ.get('TORNADO_NO_EXTENSION') or
os.environ.get('TORNADO_EXTENSION') == '0'):
projects / thirdparty / tornado.git / commitdiff
