colorama = None
try:
- import curses # type: ignore
+ import curses
except ImportError:
curses = None # type: ignore
+from typing import Dict, Any, cast
+
# Logger objects for internal tornado use
access_log = logging.getLogger("tornado.access")
app_log = logging.getLogger("tornado.application")
gen_log = logging.getLogger("tornado.general")
-def _stderr_supports_color():
+def _stderr_supports_color() -> bool:
try:
if hasattr(sys.stderr, 'isatty') and sys.stderr.isatty():
if curses:
return False
-def _safe_unicode(s):
+def _safe_unicode(s: Any) -> str:
try:
return _unicode(s)
except UnicodeDecodeError:
logging.ERROR: 1, # Red
}
- def __init__(self, fmt=DEFAULT_FORMAT, datefmt=DEFAULT_DATE_FORMAT,
- style='%', color=True, colors=DEFAULT_COLORS):
+ def __init__(self, fmt: str=DEFAULT_FORMAT, datefmt: str=DEFAULT_DATE_FORMAT,
+ style: str='%', color: bool=True, colors: Dict[int, int]=DEFAULT_COLORS) -> None:
r"""
:arg bool color: Enables color support.
:arg str fmt: Log message format.
logging.Formatter.__init__(self, datefmt=datefmt)
self._fmt = fmt
- self._colors = {}
+ self._colors = {} # type: Dict[int, str]
if color and _stderr_supports_color():
if curses is not None:
- # The curses module has some str/bytes confusion in
- # python3. Until version 3.2.3, most methods return
- # bytes, but only accept strings. In addition, we want to
- # output these strings with the logging module, which
- # works with unicode strings. The explicit calls to
- # unicode() below are harmless in python2 but will do the
- # right conversion in python 3.
fg_color = (curses.tigetstr("setaf") or
- curses.tigetstr("setf") or "")
- if (3, 0) < sys.version_info < (3, 2, 3):
- fg_color = unicode_type(fg_color, "ascii")
+ curses.tigetstr("setf") or b"")
for levelno, code in colors.items():
+ # Convert the terminal control characters from
+ # bytes to unicode strings for easier use with the
+ # logging module.
self._colors[levelno] = unicode_type(curses.tparm(fg_color, code), "ascii")
self._normal = unicode_type(curses.tigetstr("sgr0"), "ascii")
else:
else:
self._normal = ''
- def format(self, record):
+ def format(self, record: Any) -> str:
try:
message = record.getMessage()
assert isinstance(message, basestring_type) # guaranteed by logging
except Exception as e:
record.message = "Bad message (%r): %r" % (e, record.__dict__)
- record.asctime = self.formatTime(record, self.datefmt)
+ record.asctime = self.formatTime(record, cast(str, self.datefmt))
if record.levelno in self._colors:
record.color = self._colors[record.levelno]
return formatted.replace("\n", "\n ")
-def enable_pretty_logging(options=None, logger=None):
+def enable_pretty_logging(options: Any=None,
+ logger: logging.Logger=None) -> None:
"""Turns on formatted logging output as configured.
This is called automatically by `tornado.options.parse_command_line`
channel = logging.handlers.RotatingFileHandler(
filename=options.log_file_prefix,
maxBytes=options.log_file_max_size,
- backupCount=options.log_file_num_backups)
+ backupCount=options.log_file_num_backups) # type: logging.Handler
elif rotate_mode == 'time':
channel = logging.handlers.TimedRotatingFileHandler(
filename=options.log_file_prefix,
logger.addHandler(channel)
-def define_logging_options(options=None):
+def define_logging_options(options: Any=None) -> None:
"""Add logging-related flags to ``options``.
These options are present automatically on the default options instance;
from tornado.log import define_logging_options
from tornado.util import basestring_type, exec_in
+import typing
+from typing import Any, Iterator, Iterable, Tuple, Set, Dict, Callable, List, TextIO
+
+if typing.TYPE_CHECKING:
+ from typing import Optional # noqa: F401
+
class Error(Exception):
"""Exception raised by errors in the options module."""
Normally accessed via static functions in the `tornado.options` module,
which reference a global instance.
"""
- def __init__(self):
+ def __init__(self) -> None:
# we have to use self.__dict__ because we override setattr.
self.__dict__['_options'] = {}
self.__dict__['_parse_callbacks'] = []
self.define("help", type=bool, help="show this help information",
callback=self._help_callback)
- def _normalize_name(self, name):
+ def _normalize_name(self, name: str) -> str:
return name.replace('_', '-')
- def __getattr__(self, name):
+ def __getattr__(self, name: str) -> Any:
name = self._normalize_name(name)
if isinstance(self._options.get(name), _Option):
return self._options[name].value()
raise AttributeError("Unrecognized option %r" % name)
- def __setattr__(self, name, value):
+ def __setattr__(self, name: str, value: Any) -> None:
name = self._normalize_name(name)
if isinstance(self._options.get(name), _Option):
return self._options[name].set(value)
raise AttributeError("Unrecognized option %r" % name)
- def __iter__(self):
+ def __iter__(self) -> Iterator:
return (opt.name for opt in self._options.values())
- def __contains__(self, name):
+ def __contains__(self, name: str) -> bool:
name = self._normalize_name(name)
return name in self._options
- def __getitem__(self, name):
+ def __getitem__(self, name: str) -> Any:
return self.__getattr__(name)
- def __setitem__(self, name, value):
+ def __setitem__(self, name: str, value: Any) -> None:
return self.__setattr__(name, value)
- def items(self):
- """A sequence of (name, value) pairs.
+ def items(self) -> Iterable[Tuple[str, Any]]:
+ """An iterable of (name, value) pairs.
.. versionadded:: 3.1
"""
return [(opt.name, opt.value()) for name, opt in self._options.items()]
- def groups(self):
+ def groups(self) -> Set[str]:
"""The set of option-groups created by ``define``.
.. versionadded:: 3.1
"""
return set(opt.group_name for opt in self._options.values())
- def group_dict(self, group):
+ def group_dict(self, group: str) -> Dict[str, Any]:
"""The names and values of options in a group.
Useful for copying options into Application settings::
(opt.name, opt.value()) for name, opt in self._options.items()
if not group or group == opt.group_name)
- def as_dict(self):
+ def as_dict(self) -> Dict[str, Any]:
"""The names and values of all options.
.. versionadded:: 3.1
return dict(
(opt.name, opt.value()) for name, opt in self._options.items())
- def define(self, name, default=None, type=None, help=None, metavar=None,
- multiple=False, group=None, callback=None):
+ def define(self, name: str, default: Any=None, type: type=None,
+ help: str=None, metavar: str=None,
+ multiple: bool=False, group: str=None, callback: Callable[[Any], None]=None) -> None:
"""Defines a new command line option.
``type`` can be any of `str`, `int`, `float`, `bool`,
else:
type = str
if group:
- group_name = group
+ group_name = group # type: Optional[str]
else:
group_name = file_name
option = _Option(name, file_name=file_name,
callback=callback)
self._options[normalized] = option
- def parse_command_line(self, args=None, final=True):
+ def parse_command_line(self, args: List[str]=None, final: bool=True) -> List[str]:
"""Parses all options given on the command line (defaults to
`sys.argv`).
"""
if args is None:
args = sys.argv
- remaining = []
+ remaining = [] # type: List[str]
for i in range(1, len(args)):
# All things after the last option are command line arguments
if not args[i].startswith("-"):
return remaining
- def parse_config_file(self, path, final=True):
+ def parse_config_file(self, path: str, final: bool=True) -> None:
"""Parses and loads the config file at the given path.
The config file contains Python code that will be executed (so
if final:
self.run_parse_callbacks()
- def print_help(self, file=None):
+ def print_help(self, file: TextIO=None) -> None:
"""Prints all the command line options to stderr (or another file)."""
if file is None:
file = sys.stderr
print("Usage: %s [OPTIONS]" % sys.argv[0], file=file)
print("\nOptions:\n", file=file)
- by_group = {}
+ by_group = {} # type: Dict[str, List[_Option]]
for option in self._options.values():
by_group.setdefault(option.group_name, []).append(option)
print("%-34s %s" % (' ', line), file=file)
print(file=file)
- def _help_callback(self, value):
+ def _help_callback(self, value: bool) -> None:
if value:
self.print_help()
sys.exit(0)
- def add_parse_callback(self, callback):
+ def add_parse_callback(self, callback: Callable[[], None]) -> None:
"""Adds a parse callback, to be invoked when option parsing is done."""
self._parse_callbacks.append(callback)
- def run_parse_callbacks(self):
+ def run_parse_callbacks(self) -> None:
for callback in self._parse_callbacks:
callback()
- def mockable(self):
+ def mockable(self) -> '_Mockable':
"""Returns a wrapper around self that is compatible with
`mock.patch <unittest.mock.patch>`.
_Mockable's getattr and setattr pass through to the underlying
OptionParser, and delattr undoes the effect of a previous setattr.
"""
- def __init__(self, options):
+ def __init__(self, options: OptionParser) -> None:
# Modify __dict__ directly to bypass __setattr__
self.__dict__['_options'] = options
self.__dict__['_originals'] = {}
- def __getattr__(self, name):
+ def __getattr__(self, name: str) -> Any:
return getattr(self._options, name)
- def __setattr__(self, name, value):
+ def __setattr__(self, name: str, value: Any) -> None:
assert name not in self._originals, "don't reuse mockable objects"
self._originals[name] = getattr(self._options, name)
setattr(self._options, name, value)
- def __delattr__(self, name):
+ def __delattr__(self, name: str) -> None:
setattr(self._options, name, self._originals.pop(name))
class _Option(object):
+ # This class could almost be made generic, but the way the types
+ # interact with the multiple argument makes this tricky. (default
+ # and the callback use List[T], but type is still Type[T]).
UNSET = object()
- def __init__(self, name, default=None, type=basestring_type, help=None,
- metavar=None, multiple=False, file_name=None, group_name=None,
- callback=None):
+ def __init__(self, name: str, default: Any=None, type: type=None,
+ help: str=None, metavar: str=None, multiple: bool=False, file_name: str=None,
+ group_name: str=None, callback: Callable[[Any], None]=None) -> None:
if default is None and multiple:
default = []
self.name = name
+ if type is None:
+ raise ValueError("type must not be None")
self.type = type
self.help = help
self.metavar = metavar
self.group_name = group_name
self.callback = callback
self.default = default
- self._value = _Option.UNSET
+ self._value = _Option.UNSET # type: Any
- def value(self):
+ def value(self) -> Any:
return self.default if self._value is _Option.UNSET else self._value
- def parse(self, value):
+ def parse(self, value: str) -> Any:
_parse = {
datetime.datetime: self._parse_datetime,
datetime.timedelta: self._parse_timedelta,
bool: self._parse_bool,
basestring_type: self._parse_string,
- }.get(self.type, self.type)
+ }.get(self.type, self.type) # type: Callable[[str], Any]
if self.multiple:
self._value = []
for part in value.split(","):
if issubclass(self.type, numbers.Integral):
# allow ranges of the form X:Y (inclusive at both ends)
- lo, _, hi = part.partition(":")
- lo = _parse(lo)
- hi = _parse(hi) if hi else lo
+ lo_str, _, hi_str = part.partition(":")
+ lo = _parse(lo_str)
+ hi = _parse(hi_str) if hi_str else lo
self._value.extend(range(lo, hi + 1))
else:
self._value.append(_parse(part))
self.callback(self._value)
return self.value()
- def set(self, value):
+ def set(self, value: Any) -> None:
if self.multiple:
if not isinstance(value, list):
raise Error("Option %r is required to be a list of %s" %
"%H:%M",
]
- def _parse_datetime(self, value):
+ def _parse_datetime(self, value: str) -> datetime.datetime:
for format in self._DATETIME_FORMATS:
try:
return datetime.datetime.strptime(value, format)
_TIMEDELTA_PATTERN = re.compile(
r'\s*(%s)\s*(\w*)\s*' % _FLOAT_PATTERN, re.IGNORECASE)
- def _parse_timedelta(self, value):
+ def _parse_timedelta(self, value: str) -> datetime.timedelta:
try:
sum = datetime.timedelta()
start = 0
except Exception:
raise
- def _parse_bool(self, value):
+ def _parse_bool(self, value: str) -> bool:
return value.lower() not in ("false", "0", "f")
- def _parse_string(self, value):
+ def _parse_string(self, value: str) -> str:
return _unicode(value)
"""
-def define(name, default=None, type=None, help=None, metavar=None,
- multiple=False, group=None, callback=None):
+def define(name: str, default: Any=None, type: type=None, help: str=None,
+ metavar: str=None, multiple: bool=False, group: str=None,
+ callback: Callable[[Any], None]=None) -> None:
"""Defines an option in the global namespace.
See `OptionParser.define`.
callback=callback)
-def parse_command_line(args=None, final=True):
+def parse_command_line(args: List[str]=None, final: bool=True) -> List[str]:
"""Parses global options from the command line.
See `OptionParser.parse_command_line`.
return options.parse_command_line(args, final=final)
-def parse_config_file(path, final=True):
+def parse_config_file(path: str, final: bool=True) -> None:
"""Parses global options from a config file.
See `OptionParser.parse_config_file`.
return options.parse_config_file(path, final=final)
-def print_help(file=None):
+def print_help(file: TextIO=None) -> None:
"""Prints all the command line options to stderr (or another file).
See `OptionParser.print_help`.
return options.print_help(file)
-def add_parse_callback(callback):
+def add_parse_callback(callback: Callable[[], None]) -> None:
"""Adds a parse callback, to be invoked when option parsing is done.
See `OptionParser.add_parse_callback`
projects / thirdparty / tornado.git / commitdiff
