# version supported. But Python 3.7 requires slower-to-start VMs,
# so we run it on 3.6 to minimize total CI run time.
- if [[ $TRAVIS_PYTHON_VERSION == '3.6' ]]; then travis_retry pip install sphinx sphinx_rtd_theme; fi
- - if [[ $TRAVIS_PYTHON_VERSION == '3.6' ]]; then travis_retry pip install flake8; fi
+ - if [[ $TRAVIS_PYTHON_VERSION == '3.6' ]]; then travis_retry pip install flake8 mypy; fi
# On travis the extension should always be built
- if [[ $TRAVIS_PYTHON_VERSION != 'pypy'* ]]; then export TORNADO_EXTENSION=1; fi
- travis_retry python setup.py install
# make coverage reports for Codecov to find
- if [[ "$RUN_COVERAGE" == 1 ]]; then coverage xml; fi
- export TORNADO_EXTENSION=0
- - if [[ $TRAVIS_PYTHON_VERSION == 3.6 ]]; then cd ../docs && mkdir sphinx-out && sphinx-build -E -n -W -b html . sphinx-out; fi
- - if [[ $TRAVIS_PYTHON_VERSION == 3.6 ]]; then cd ../docs && mkdir sphinx-doctest-out && sphinx-build -E -n -b doctest . sphinx-out; fi
- - if [[ $TRAVIS_PYTHON_VERSION == '3.6' ]]; then cd .. && flake8; fi
+ - if [[ $TRAVIS_PYTHON_VERSION == 3.6 ]]; then (cd ../docs && mkdir sphinx-out && sphinx-build -E -n -W -b html . sphinx-out); fi
+ - if [[ $TRAVIS_PYTHON_VERSION == 3.6 ]]; then (cd ../docs && mkdir sphinx-doctest-out && sphinx-build -E -n -b doctest . sphinx-out); fi
+ - if [[ $TRAVIS_PYTHON_VERSION == '3.6' ]]; then (cd .. && flake8); fi
+ - if [[ $TRAVIS_PYTHON_VERSION == '3.6' ]]; then (cd .. && mypy tornado); fi
after_success:
# call codecov from project root
try:
import signal
except ImportError:
- signal = None
+ signal = None # type: ignore
# os.execv is broken on Windows and can't properly parse command line
# arguments and executable name if they contain whitespaces. subprocess
from tornado.log import app_log
from tornado.util import Configurable, TimeoutError, unicode_type, import_object
+import typing # noqa
+
class IOLoop(Configurable):
"""A level-triggered I/O loop.
ERROR = 0x018
# In Python 3, _ioloop_for_asyncio maps from asyncio loops to IOLoops.
- _ioloop_for_asyncio = dict()
+ _ioloop_for_asyncio = dict() # type: typing.Dict[typing.Any, typing.Any]
@classmethod
def configure(cls, impl, **kwargs):
try:
from tornado.platform.posix import _set_nonblocking
except ImportError:
- _set_nonblocking = None
+ _set_nonblocking = None # type: ignore
# These errnos indicate that a non-blocking operation must be retried
# at a later time. On most platforms they're the same value, but on
from tornado.util import unicode_type, basestring_type
try:
- import colorama
+ import colorama # type: ignore
except ImportError:
colorama = None
try:
import curses # type: ignore
except ImportError:
- curses = None
+ curses = None # type: ignore
# Logger objects for internal tornado use
access_log = logging.getLogger("tornado.access")
return convert_yielded(tornado_future)
-class AnyThreadEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
+class AnyThreadEventLoopPolicy(asyncio.DefaultEventLoopPolicy): # type: ignore
"""Event loop policy that allows loop creation on any thread.
The default `asyncio` event loop policy only automatically creates
# NOTE: win32 support is currently experimental, and not recommended
# for production use.
-import ctypes # type: ignore
-import ctypes.wintypes # type: ignore
+import ctypes
+import ctypes.wintypes
# See: http://msdn.microsoft.com/en-us/library/ms724935(VS.85).aspx
-SetHandleInformation = ctypes.windll.kernel32.SetHandleInformation
+SetHandleInformation = ctypes.windll.kernel32.SetHandleInformation # type: ignore
SetHandleInformation.argtypes = (ctypes.wintypes.HANDLE, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD) # noqa: E501
SetHandleInformation.restype = ctypes.wintypes.BOOL
"""
def __init__(self, rules=None):
- self.named_rules = {} # type: typing.Dict[str]
+ self.named_rules = {} # type: typing.Dict[str, Any]
super(ReversibleRuleRouter, self).__init__(rules)
def process_rule(self, rule):
class AllMethodsHandler(RequestHandler):
- SUPPORTED_METHODS = RequestHandler.SUPPORTED_METHODS + ('OTHER',)
+ SUPPORTED_METHODS = RequestHandler.SUPPORTED_METHODS + ('OTHER',) # type: ignore
def method(self):
self.write(self.request.method)
from tornado.web import Application, HTTPError, RequestHandler
from tornado.wsgi import WSGIContainer
+import typing # noqa
+
class BasicRouter(Router):
def find_handler(self, request, **kwargs):
self.assertEqual(response.body, b"OK")
-resources = {}
+resources = {} # type: typing.Dict[str, bytes]
class GetResource(RequestHandler):
try:
from twisted.internet.defer import Deferred, inlineCallbacks, returnValue # type: ignore
from twisted.internet.protocol import Protocol # type: ignore
- from twisted.internet.asyncioreactor import AsyncioSelectorReactor
+ from twisted.internet.asyncioreactor import AsyncioSelectorReactor # type: ignore
from twisted.web.client import Agent, readBody # type: ignore
from twisted.web.resource import Resource # type: ignore
from twisted.web.server import Site # type: ignore
# TODO: remove this after pypy3 5.8 is obsolete.
skipPypy3V58 = unittest.skipIf(platform.python_implementation() == 'PyPy' and
sys.version_info > (3,) and
- sys.pypy_version_info < (5, 9),
+ sys.pypy_version_info < (5, 9), # type: ignore
'pypy3 5.8 has buggy ssl module')
class PatchMethodTest(SimpleHandlerTestCase):
class Handler(RequestHandler):
- SUPPORTED_METHODS = RequestHandler.SUPPORTED_METHODS + ('OTHER',)
+ SUPPORTED_METHODS = RequestHandler.SUPPORTED_METHODS + ('OTHER',) # type: ignore
def patch(self):
self.write('patch')
try:
from tornado import speedups
except ImportError:
- speedups = None
+ speedups = None # type: ignore
class TestWebSocketHandler(WebSocketHandler):
unicode_type = str
basestring_type = str
-try:
- import typing # noqa
- from typing import cast
-
- _ObjectDictBase = typing.Dict[str, typing.Any]
-except ImportError:
- _ObjectDictBase = dict
+import typing
- 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
+# More imports that are only needed in type annotations.
+import datetime # noqa
+import types
+from typing import Any, AnyStr, Union, Optional, Dict, Mapping, List # noqa
+from typing import Tuple, Match, Callable # noqa
- _BaseString = str
+_BaseString = str
try:
from sys import is_finalizing
"""
-class ObjectDict(_ObjectDictBase):
+class ObjectDict(typing.Dict[str, typing.Any]):
"""Makes a dictionary behave like an object, with attribute-style access.
"""
def __getattr__(self, name):
# on python 2 a byte string is required.
name = name.encode('utf-8')
if name.count('.') == 0:
- return __import__(name, None, None)
+ return __import__(name)
parts = name.split('.')
- obj = __import__('.'.join(parts[:-1]), None, None, [parts[-1]], 0)
+ obj = __import__('.'.join(parts[:-1]), fromlist=[parts[-1]])
try:
return getattr(obj, parts[-1])
except AttributeError:
exec(code, glob, loc)
-def raise_exc_info(exc_info):
- # type: (Tuple[type, BaseException, types.TracebackType]) -> None
+def raise_exc_info(exc_info: Optional[Tuple[type, BaseException, types.TracebackType]]) -> None:
try:
- raise exc_info[1].with_traceback(exc_info[2])
+ if exc_info is not None:
+ raise exc_info[1].with_traceback(exc_info[2])
finally:
exc_info = None
# type: (Callable, str) -> None
self.name = name
try:
- self.arg_pos = self._getargnames(func).index(name)
+ self.arg_pos = self._getargnames(func).index(name) # type: Optional[int]
except ValueError:
# Not a positional parameter
self.arg_pos = None
py3-sphinx-doctest,
py3-lint
+ py3-mypy
[testenv]
# Most of these are defaults, but if you specify any you can't fall back
[testenv:py3-lint]
commands = flake8 {posargs:}
changedir = {toxinidir}
+
+[testenv:py3-mypy]
+commands = mypy {posargs:tornado}
+changedir = {toxinidir}
+deps = mypy