from tornado.concurrent import Future, is_future, chain_future, future_set_exc_info, future_add_done_callback # noqa: E501
from tornado.log import app_log
from tornado import stack_context
-from tornado.util import Configurable, timedelta_to_seconds, TimeoutError, unicode_type, import_object
+from tornado.util import Configurable, TimeoutError, unicode_type, import_object
class IOLoop(Configurable):
if isinstance(deadline, numbers.Real):
return self.call_at(deadline, callback, *args, **kwargs)
elif isinstance(deadline, datetime.timedelta):
- return self.call_at(self.time() + timedelta_to_seconds(deadline),
+ return self.call_at(self.time() + deadline.total_seconds(),
callback, *args, **kwargs)
else:
raise TypeError("Unsupported deadline %r" % deadline)
from tornado.netutil import Resolver
from tornado.platform.auto import set_close_exec
from tornado.gen import TimeoutError
-from tornado.util import timedelta_to_seconds
_INITIAL_CONNECT_TIMEOUT = 0.3
if isinstance(timeout, numbers.Real):
timeout = IOLoop.current().time() + timeout
elif isinstance(timeout, datetime.timedelta):
- timeout = IOLoop.current().time() + timedelta_to_seconds(timeout)
+ timeout = IOLoop.current().time() + timeout.total_seconds()
else:
raise TypeError("Unsupported timeout %r" % timeout)
if timeout is not None:
from tornado.template import DictLoader
from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, ExpectLog, gen_test
from tornado.test.util import unittest, skipBefore35, exec_test, ignore_deprecation
-from tornado.util import ObjectDict, unicode_type, timedelta_to_seconds, PY3
+from tornado.util import ObjectDict, unicode_type, PY3
from tornado.web import (
Application, RequestHandler, StaticFileHandler, RedirectHandler as WebRedirectHandler,
HTTPError, MissingArgumentError, ErrorHandler, authenticated, asynchronous, url,
expires = datetime.datetime.utcnow() + datetime.timedelta(days=10)
header_expires = datetime.datetime(
*email.utils.parsedate(match.groupdict()["expires"])[:6])
- self.assertTrue(abs(timedelta_to_seconds(expires - header_expires)) < 10)
+ self.assertTrue(abs((expires - header_expires).total_seconds()) < 10)
def test_set_cookie_false_flags(self):
response = self.fetch("/set_falsy_flags")
import array
import atexit
+from inspect import getfullargspec
import os
import re
import sys
PY3 = sys.version_info >= (3,)
-if PY3:
- xrange = range
-
-# inspect.getargspec() raises DeprecationWarnings in Python 3.5.
-# The two functions have compatible interfaces for the parts we need.
-if PY3:
- from inspect import getfullargspec as getargspec
-else:
- from inspect import getargspec
-
# Aliases for types that are spelled differently in different Python
# versions. bytes_type is deprecated and no longer used in Tornado
# itself but is left in case anyone outside Tornado is using it.
bytes_type = bytes
-if PY3:
- unicode_type = str
- basestring_type = str
-else:
- # The names unicode and basestring don't exist in py3 so silence flake8.
- unicode_type = unicode # noqa
- basestring_type = basestring # noqa
-
+unicode_type = str
+basestring_type = str
try:
import typing # noqa
raise ImportError("No module named %s" % parts[-1])
-# 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
if isinstance(code, basestring_type):
exec(code, glob, loc)
-if PY3:
- exec("""
def raise_exc_info(exc_info):
+ # type: (Tuple[type, BaseException, types.TracebackType]) -> None
try:
raise exc_info[1].with_traceback(exc_info[2])
finally:
exc_info = None
-""")
-else:
- exec("""
-def raise_exc_info(exc_info):
- raise exc_info[0], exc_info[1], exc_info[2]
-""")
-
def errno_from_exception(e):
# type: (BaseException) -> Optional[int]
def _getargnames(self, func):
# type: (Callable) -> List[str]
try:
- return getargspec(func).args
+ return getfullargspec(func).args
except TypeError:
if hasattr(func, 'func_code'):
# Cython-generated code has all the attributes needed
- # by inspect.getargspec, but the inspect module only
+ # by inspect.getfullargspec, but the inspect module only
# works with ordinary functions. Inline the portion of
- # getargspec that we need here. Note that for static
+ # getfullargspec 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 # type: ignore
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)
+ return td.total_seconds()
def _websocket_mask_python(mask, data):
"""
mask_arr = array.array("B", mask)
unmasked_arr = array.array("B", data)
- for i in xrange(len(data)):
+ for i in range(len(data)):
unmasked_arr[i] = unmasked_arr[i] ^ mask_arr[i % 4]
if PY3:
# tostring was deprecated in py32. It hasn't been removed,