Merge pull request #3826 from jnoortheen/feat-add-mypy

feat: add mypy to project

.gitignore

@@ -80,3 +80,6 @@ venv/
 
 # AppImage
 *.AppImage
+
+# mypy
+.dmypy.json

ci/environment-3.5.yml

@@ -1,17 +0,0 @@
-name: 3.5-xonsh-test
-channels:
-  - conda-forge
-  - defaults
-dependencies:
-  - python=3.5
-  - pygments
-  - prompt_toolkit
-  - pytest
-  - pytest-timeout
-  - numpy
-  - psutil
-  - flake8
-  - coverage
-  - pyflakes
-  - pytest-cov
-  - codecov

ci/environment-3.6.yml

@@ -12,6 +12,7 @@ dependencies:
   - psutil
   - matplotlib
   - flake8
+  - mypy=0.782
   - coverage
   - pyflakes
   - pytest-cov

ci/environment-3.7.yml

@@ -12,6 +12,7 @@ dependencies:
   - psutil
   - matplotlib
   - flake8
+  - mypy=0.782
   - coverage
   - pyflakes
   - pytest-cov

ci/environment-3.8.yml

@@ -12,6 +12,7 @@ dependencies:
   - psutil
   - matplotlib
   - flake8
+  - mypy=0.782
   - coverage
   - pyflakes
   - pytest-cov

news/feat-add-mypy.rst

@@ -0,0 +1,23 @@
+**Added:**
+
+* added mypy to the project. many of the errors are ignored. but it is a start.
+
+**Changed:**
+
+* <news item>
+
+**Deprecated:**
+
+* <news item>
+
+**Removed:**
+
+* <news item>
+
+**Fixed:**
+
+* <news item>
+
+**Security:**
+
+* <news item>

requirements/development.txt

@@ -0,0 +1,3 @@
+-r tests.txt
+sniffer
+pyinotify

requirements/tests-35.txt

@@ -1,9 +0,0 @@
-py
-pytest>=5.4
-flake8
-pytest-cov
-pytest-timeout
-prompt-toolkit>=2.0
-pygments>=2.2
-codecov
-coverage

requirements/tests.txt

@@ -8,5 +8,6 @@ prompt-toolkit>=2.0
 pygments>=2.2
 codecov
 coverage
-black
+black==19.10b0 --pre
 pre-commit
+mypy>=0.782

run-tests.xsh

@@ -30,3 +30,6 @@ for index, fname in enumerate(run_separately):
 echo "---------- Running flake8 ----------"
 python -m flake8
     
+echo "---------- Running mypy ----------"
+mypy --version
+mypy xonsh

scent.py

@@ -0,0 +1,49 @@
+"""Configuration file for sniffer."""
+
+import time
+import subprocess
+
+from sniffer.api import select_runnable, file_validator, runnable
+
+try:
+    from pync import Notifier
+except ImportError:
+    notify = None
+else:
+    notify = Notifier.notify
+
+watch_paths = ['.']
+
+
+@select_runnable('python')
+@file_validator
+def py_files(filename):
+    return "TemplateDemo" not in filename
+
+
+@runnable
+def python(*_):
+    group = int(time.time())  # unique per run
+
+    for count, (command, title) in enumerate((
+            (('dmypy', 'run', "--", "xonsh"), "Lint"),
+            (('flake8', '--count'), "Lint"),
+            (('pytest', 'tests/test_main.py'), "Test main"),
+            (('pytest', 'tests/test_ptk_highlight.py'), "Test ptk highlight"),
+            (('pytest', '--ignore', 'tests/test_main.py', 'tests/test_ptk_highlight.py'), "Test Rest"),
+    ), start=1):
+
+        print(f"\n$ {' '.join(command)}")
+        failure = subprocess.call(command)
+
+        if failure:
+            if notify and title:
+                mark = "❌" * count
+                notify(mark + " [FAIL] " + mark, title=title, group=group)
+            return False
+        else:
+            if notify and title:
+                mark = "✅" * count
+                notify(mark + " [PASS] " + mark, title=title, group=group)
+
+    return True

setup.cfg

@@ -97,3 +97,29 @@ per-file-ignores =
 
 # pydocstyle plugin
 docstring-convention=numpy
+
+[mypy]
+# --- https://mypy.readthedocs.io/en/stable/config_file.html
+# try to keep all under .cache directory
+cache_dir = .cache/mypy/
+# warn_unused_ignores = True
+warn_unused_configs = True
+warn_no_return = False
+
+# report
+show_error_context = True
+show_column_numbers = True
+show_error_codes = True
+pretty = True
+
+# the __init__ files have dynamic check - ignoring the attribute error. others are generated files
+# top level package name only ignores the __init__.py file.
+[mypy-xonsh.parser_table,xonsh.parsers.parser_table.*,*.__amalgam__.*,xonsh,xonsh.prompt,xonsh.history,xonsh.completers]
+ignore_errors = True
+
+# 3rd party libraries that we dont have control over
+[mypy-zmq.*,setproctitle,xonsh.ply.*,jupyter_client.*,winreg.*,pygments.*,prompt_toolkit.*,importlib_resources.*,nt.*,prompt_toolkit.*,distro.*,conda_suggest.*,_winreg.*]
+ignore_missing_imports = True
+
+[tool:pytest]
+cache_dir = .cache/pytest

xonsh/ansi_colors.py

@@ -3,6 +3,7 @@ import re
 import sys
 import warnings
 import builtins
+import typing as tp
 
 from xonsh.platform import HAS_PYGMENTS
 from xonsh.lazyasd import LazyDict, lazyobject
@@ -261,7 +262,7 @@ def _color_name_from_ints(ints, background=False, prefix=None):
     return name
 
 
-_ANSI_COLOR_ESCAPE_CODE_TO_NAME_CACHE = {}
+_ANSI_COLOR_ESCAPE_CODE_TO_NAME_CACHE: tp.Dict[str, tp.Tuple[str, ...]] = {}
 
 
 def ansi_color_escape_code_to_name(escape_code, style, reversed_style=None):
@@ -274,6 +275,7 @@ def ansi_color_escape_code_to_name(escape_code, style, reversed_style=None):
     style is not provided, it is computed.
     """
     key = (escape_code, style)
+    # todo: see the cache ever used?
     if key in _ANSI_COLOR_ESCAPE_CODE_TO_NAME_CACHE:
         return _ANSI_COLOR_ESCAPE_CODE_TO_NAME_CACHE[key]
     if reversed_style is None:

xonsh/ast.py

@@ -125,7 +125,7 @@ from ast import (
 from xonsh.platform import PYTHON_VERSION_INFO
 
 if PYTHON_VERSION_INFO > (3, 8):
-    from ast import NamedExpr
+    from ast import NamedExpr  # type:ignore
 
 STATEMENTS = (
     FunctionDef,

xonsh/base_shell.py

@@ -33,7 +33,7 @@ from xonsh.ansi_colors import ansi_partial_color_format
 if ON_WINDOWS:
     import ctypes
 
-    kernel32 = ctypes.windll.kernel32
+    kernel32 = ctypes.windll.kernel32  # type:ignore
     kernel32.SetConsoleTitleW.argtypes = [ctypes.c_wchar_p]
 
 

xonsh/completers/bash_completion.py

@@ -13,6 +13,7 @@ import pathlib
 import platform
 import functools
 import subprocess
+import typing as tp
 
 __version__ = "0.2.6"
 
@@ -98,7 +99,7 @@ def _bash_completion_paths_default():
     return bcd
 
 
-_BASH_COMPLETIONS_PATHS_DEFAULT = None
+_BASH_COMPLETIONS_PATHS_DEFAULT: tp.Tuple[str, ...] = ()
 
 
 def _get_bash_completions_source(paths=None):
@@ -123,7 +124,7 @@ def _bash_get_sep():
         return os.sep
 
 
-_BASH_PATTERN_NEED_QUOTES = None
+_BASH_PATTERN_NEED_QUOTES: tp.Optional[tp.Pattern] = None
 
 
 def _bash_pattern_need_quotes():

xonsh/completers/man.py

@@ -3,13 +3,14 @@ import re
 import pickle
 import builtins
 import subprocess
+import typing as tp
 
 import xonsh.lazyasd as xl
 
 from xonsh.completers.tools import get_filter_function
 
-OPTIONS = None
-OPTIONS_PATH = None
+OPTIONS: tp.Optional[tp.Dict[str, tp.Any]] = None
+OPTIONS_PATH: tp.Optional[str] = None
 
 
 @xl.lazyobject

xonsh/diff_history.py

@@ -4,6 +4,7 @@ import difflib
 import datetime
 import itertools
 import argparse
+import typing as tp
 
 from xonsh.lazyjson import LazyJSON
 from xonsh.tools import print_color
@@ -302,7 +303,7 @@ class HistoryDiffer(object):
         return s.rstrip()
 
 
-_HD_PARSER = None
+_HD_PARSER: tp.Optional[argparse.ArgumentParser] = None
 
 
 def dh_create_parser(p=None):

xonsh/dirstack.py

@@ -1,19 +1,20 @@
 # -*- coding: utf-8 -*-
 """Directory stack and associated utilities for the xonsh shell."""
-import os
-import glob
 import argparse
 import builtins
+import glob
+import os
 import subprocess
+import typing as tp
 
-from xonsh.lazyasd import lazyobject
-from xonsh.tools import get_sep
 from xonsh.events import events
+from xonsh.lazyasd import lazyobject
 from xonsh.platform import ON_WINDOWS
+from xonsh.tools import get_sep
 
-DIRSTACK = []
+DIRSTACK: tp.List[str] = []
 """A list containing the currently remembered directories."""
-_unc_tempDrives = {}
+_unc_tempDrives: tp.Dict[str, str] = {}
 """ drive: sharePath for temp drive letters we create for UNC mapping"""
 
 
@@ -27,7 +28,7 @@ def _unc_check_enabled() -> bool:
         False if check is explicitly disabled.
     """
     if not ON_WINDOWS:
-        return
+        return False
 
     import winreg
 

xonsh/environ.py

@@ -13,6 +13,7 @@ import collections
 import collections.abc as cabc
 import subprocess
 import platform
+import typing as tp
 
 from xonsh import __version__ as XONSH_VERSION
 from xonsh.lazyasd import LazyObject, lazyobject
@@ -672,7 +673,7 @@ doc_store_as_str : bool, optional
 """
 
 # iterates from back
-Var.__new__.__defaults__ = (
+Var.__new__.__defaults__ = (  # type:ignore
     always_true,
     None,
     ensure_string,
@@ -1755,7 +1756,8 @@ class Env(cabc.MutableMapping):
     use in a subprocess.
     """
 
-    _arg_regex = None
+    # todo: check this variable is ever used
+    _arg_regex: tp.Optional[str] = None
 
     def __init__(self, *args, **kwargs):
         """If no initial environment is given, os_environ is used."""

xonsh/fs.py

@@ -1,98 +0,0 @@
-"""
-Backported functions to implement the PEP 519 (Adding a file system path protocol) API.
-"""
-
-import abc
-import sys
-import io
-import pathlib
-
-try:
-    from os import PathLike, fspath, fsencode, fsdecode
-except ImportError:
-
-    class PathLike(abc.ABC):
-        """Abstract base class for implementing the file system path protocol."""
-
-        @abc.abstractmethod
-        def __fspath__(self):
-            """Return the file system path representation of the object."""
-            raise NotImplementedError
-
-    PathLike.register(pathlib.Path)
-
-    def fspath(path):
-        """Return the string representation of the path.
-
-        If str or bytes is passed in, it is returned unchanged. If __fspath__()
-        returns something other than str or bytes then TypeError is raised. If
-        this function is given something that is not str, bytes, or os.PathLike
-        then TypeError is raised.
-        """
-        if isinstance(path, (str, bytes)):
-            return path
-
-        if isinstance(path, pathlib.Path):
-            return str(path)
-
-        # Work from the object's type to match method resolution of other magic
-        # methods.
-        path_type = type(path)
-        try:
-            path = path_type.__fspath__(path)
-        except AttributeError:
-            if hasattr(path_type, "__fspath__"):
-                raise
-        else:
-            if isinstance(path, (str, bytes)):
-                return path
-            else:
-                raise TypeError(
-                    "expected __fspath__() to return str or bytes, "
-                    "not " + type(path).__name__
-                )
-
-        raise TypeError(
-            "expected str, bytes or os.PathLike object, not " + path_type.__name__
-        )
-
-    def _fscodec():
-        encoding = sys.getfilesystemencoding()
-        if encoding == "mbcs":
-            errors = "strict"
-        else:
-            errors = "surrogateescape"
-
-        def fsencode(filename):
-            """Encode filename (an os.PathLike, bytes, or str) to the filesystem
-            encoding with 'surrogateescape' error handler, return bytes unchanged.
-            On Windows, use 'strict' error handler if the file system encoding is
-            'mbcs' (which is the default encoding).
-            """
-            filename = fspath(filename)  # Does type-checking of `filename`.
-            if isinstance(filename, str):
-                return filename.encode(encoding, errors)
-            else:
-                return filename
-
-        def fsdecode(filename):
-            """Decode filename (an os.PathLike, bytes, or str) from the filesystem
-            encoding with 'surrogateescape' error handler, return str unchanged. On
-            Windows, use 'strict' error handler if the file system encoding is
-            'mbcs' (which is the default encoding).
-            """
-            filename = fspath(filename)  # Does type-checking of `filename`.
-            if isinstance(filename, bytes):
-                return filename.decode(encoding, errors)
-            else:
-                return filename
-
-        return fsencode, fsdecode
-
-    fsencode, fsdecode = _fscodec()
-    del _fscodec
-
-    def open(file, *pargs, **kwargs):
-        if isinstance(file, PathLike):
-            file = fspath(file)
-        return io.open(file, *pargs, **kwargs)

xonsh/history/json.py

@@ -15,7 +15,7 @@ import xonsh.lazyjson as xlj
 import xonsh.xoreutils.uptime as uptime
 
 
-def _xhj_gc_commands_to_rmfiles(hsize, files) -> ([], int):
+def _xhj_gc_commands_to_rmfiles(hsize, files):
     """Return number of units and list of history files to remove to get under the limit,
 
     Parameters:

xonsh/jobs.py

@@ -8,6 +8,7 @@ import signal
 import builtins
 import subprocess
 import collections
+import typing as tp
 
 from xonsh.lazyasd import LazyObject
 from xonsh.platform import FD_STDERR, ON_DARWIN, ON_WINDOWS, ON_CYGWIN, ON_MSYS, LIBC
@@ -17,7 +18,7 @@ from xonsh.tools import unthreadable
 tasks = LazyObject(collections.deque, globals(), "tasks")
 # Track time stamp of last exit command, so that two consecutive attempts to
 # exit can kill all jobs and exit.
-_last_exit_time = None
+_last_exit_time: tp.Optional[float] = None
 
 
 if ON_DARWIN:
@@ -82,7 +83,7 @@ if ON_WINDOWS:
     def give_terminal_to(pgid):
         pass
 
-    def wait_for_active_job(last_task=None, backgrounded=False):
+    def wait_for_active_job(last_task=None, backgrounded=False, return_error=False):
         """
         Wait for the active job to finish, to be killed by SIGINT, or to be
         suspended by ctrl-z.
@@ -117,10 +118,15 @@ else:
     def ignore_sigtstp():
         signal.signal(signal.SIGTSTP, signal.SIG_IGN)
 
-    _shell_pgrp = os.getpgrp()
+    _shell_pgrp = os.getpgrp()  # type:ignore
 
     _block_when_giving = LazyObject(
-        lambda: (signal.SIGTTOU, signal.SIGTTIN, signal.SIGTSTP, signal.SIGCHLD),
+        lambda: (
+            signal.SIGTTOU,  # type:ignore
+            signal.SIGTTIN,  # type:ignore
+            signal.SIGTSTP,  # type:ignore
+            signal.SIGCHLD,  # type:ignore
+        ),
         globals(),
         "_block_when_giving",
     )
@@ -155,7 +161,7 @@ else:
             }
 
     else:
-        _pthread_sigmask = signal.pthread_sigmask
+        _pthread_sigmask = signal.pthread_sigmask  # type:ignore
 
     # give_terminal_to is a simplified version of:
     #    give_terminal_to from bash 4.3 source, jobs.c, line 4030

xonsh/jupyter_kernel.py

@@ -22,7 +22,6 @@ from xonsh.main import setup
 from xonsh.completer import Completer
 from xonsh.commands_cache import predict_true
 
-
 MAX_SIZE = 8388608  # 8 Mb
 DELIM = b"<IDS|MSG>"
 
@@ -468,11 +467,11 @@ if __name__ == "__main__":
         xontribs=["coreutils"],
         threadable_predictors={"git": predict_true, "man": predict_true},
     )
-    if builtins.__xonsh__.commands_cache.is_only_functional_alias("cat"):
+    if builtins.__xonsh__.commands_cache.is_only_functional_alias("cat"):  # type:ignore
         # this is needed if the underlying system doesn't have cat
         # we supply our own, because we can
-        builtins.aliases["cat"] = "xonsh-cat"
-        builtins.__xonsh__.env["PAGER"] = "xonsh-cat"
-    shell = builtins.__xonsh__.shell
+        builtins.aliases["cat"] = "xonsh-cat"  # type:ignore
+        builtins.__xonsh__.env["PAGER"] = "xonsh-cat"  # type:ignore
+    shell = builtins.__xonsh__.shell  # type:ignore
     kernel = shell.kernel = XonshKernel()
     kernel.start()

xonsh/lexer.py

@@ -8,6 +8,7 @@ import re
 
 # 'keyword' interferes with ast.keyword
 import keyword as kwmod
+import typing as tp
 
 from xonsh.ply.ply.lex import LexToken
 
@@ -409,7 +410,7 @@ def _new_token(type, value, pos):
 class Lexer(object):
     """Implements a lexer for the xonsh language."""
 
-    _tokens = None
+    _tokens: tp.Optional[tp.Tuple[str, ...]] = None
 
     def __init__(self):
         """

xonsh/lib/collections.py

@@ -4,12 +4,13 @@ import itertools
 
 from collections import ChainMap
 from collections.abc import MutableMapping, MutableSequence, MutableSet
+import typing as tp
 
 
 class ChainDBDefaultType(object):
     """Singleton for representing when no default value is given."""
 
-    __inst = None
+    __inst: tp.Optional["ChainDBDefaultType"] = None
 
     def __new__(cls):
         if ChainDBDefaultType.__inst is None:

xonsh/proc.py

@@ -2356,8 +2356,6 @@ class CommandPipeline:
             return 1
         return self.proc.returncode
 
-    rtn = returncode
-
     @property
     def args(self):
         """Arguments to the process."""

xonsh/ptk_shell/formatter.py

@@ -20,6 +20,7 @@ class PTKPromptFormatter(PromptFormatter):
         fields=None,
         threaded=False,
         prompt_name: str = None,
+        **_
     ) -> str:
         """Formats a xonsh prompt template string."""
 

xonsh/ptk_shell/shell.py

@@ -266,7 +266,7 @@ class PromptToolkitShell(BaseShell):
                     break
 
     def _get_prompt_tokens(self, env_name: str, prompt_name: str, **kwargs):
-        env = builtins.__xonsh__.env
+        env = builtins.__xonsh__.env  # type:ignore
         p = env.get(env_name)
 
         if not p and "default" in kwargs:

xonsh/ptk_shell/updator.py

@@ -3,7 +3,7 @@
 import builtins
 import concurrent.futures
 import threading
-from typing import Dict, List, Union, Callable, Optional
+import typing as tp
 
 from prompt_toolkit import PromptSession
 from prompt_toolkit.formatted_text import PygmentsTokens
@@ -22,7 +22,7 @@ class Executor:
         # This caches results from callback alone by field name.
         self.thread_results = {}
 
-    def submit(self, func: Callable, field: str):
+    def submit(self, func: tp.Callable, field: str):
         future = self.thread_pool.submit(self._run_func, func, field)
         place_holder = "{" + field + "}"
 
@@ -62,13 +62,13 @@ class AsyncPrompt:
         self.name = name
 
         # list of tokens in that prompt. It could either be resolved or not resolved.
-        self.tokens: List[str] = []
+        self.tokens: tp.List[str] = []
         self.timer = None
         self.session = session
         self.executor = executor
 
         # (Key: the future object) that is created for the (value: index/field_name) in the tokens list
-        self.futures: Dict[concurrent.futures.Future, Union[int, str]] = {}
+        self.futures: tp.Dict[concurrent.futures.Future, tp.Union[int, str]] = {}
 
     def start_update(self, on_complete):
         """Listen on futures and update the prompt as each one completed.
@@ -129,7 +129,7 @@ class AsyncPrompt:
             fut.cancel()
         self.futures.clear()
 
-    def submit_section(self, func: Callable, field: str, idx: int = None):
+    def submit_section(self, func: tp.Callable, field: str, idx: int = None):
         future, intermediate_value, placeholder = self.executor.submit(func, field)
         self.futures[future] = placeholder if idx is None else idx
         return intermediate_value
@@ -139,11 +139,11 @@ class PromptUpdator:
     """Handle updating multiple AsyncPrompt instances prompt/rprompt/bottom_toolbar"""
 
     def __init__(self, session: PromptSession):
-        self.prompts: Dict[str, AsyncPrompt] = {}
+        self.prompts: tp.Dict[str, AsyncPrompt] = {}
         self.prompter = session
         self.executor = Executor()
 
-    def add(self, prompt_name: Optional[str]):
+    def add(self, prompt_name: tp.Optional[str]):
         # clear out old futures from the same prompt
         if prompt_name is None:
             return
@@ -173,6 +173,6 @@ class PromptUpdator:
     def on_complete(self, prompt_name):
         self.prompts.pop(prompt_name, None)
 
-    def set_tokens(self, prompt_name, tokens: List[str]):
+    def set_tokens(self, prompt_name, tokens: tp.List[str]):
         if prompt_name in self.prompts:
             self.prompts[prompt_name].tokens = tokens

xonsh/pyghooks.py

@@ -8,6 +8,8 @@ import stat
 from collections import ChainMap
 from collections.abc import MutableMapping
 from keyword import iskeyword
+import typing as tp
+
 from xonsh.lazyimps import os_listxattr
 
 from pygments.lexer import inherit, bygroups, include
@@ -25,6 +27,7 @@ from pygments.token import (
     Token,
     Punctuation,
     Text,
+    _TokenType,
 )
 from pygments.style import Style
 import pygments.util
@@ -215,11 +218,11 @@ def code_by_name(name, styles):
     return code
 
 
-def color_token_by_name(xc: tuple, styles=None) -> Color:
+def color_token_by_name(xc: tuple, styles=None) -> _TokenType:
     """Returns (color) token corresponding to Xonsh color tuple, side effect: defines token is defined in styles"""
     if not styles:
         try:
-            styles = builtins.__xonsh__.shell.shell.styler.styles
+            styles = builtins.__xonsh__.shell.shell.styler.styles  # type:ignore
         except AttributeError:
             return Color
 
@@ -1343,7 +1346,7 @@ color_file_extension_RE = LazyObject(
 )
 
 
-file_color_tokens = dict()
+file_color_tokens = dict()  # type:ignore
 """Parallel to LS_COLORS, keyed by dircolors keys, but value is a Color token.
 Initialized by XonshStyle."""
 
@@ -1360,7 +1363,7 @@ def on_lscolors_change(key, oldvalue, newvalue, **kwargs):
 events.on_lscolors_change(on_lscolors_change)
 
 
-def color_file(file_path: str, path_stat: os.stat_result) -> (Color, str):
+def color_file(file_path: str, path_stat: os.stat_result) -> tp.Tuple[_TokenType, str]:
     """Determine color to use for file *approximately* as ls --color would,
        given lstat() results and its path.
 
@@ -1388,7 +1391,7 @@ def color_file(file_path: str, path_stat: os.stat_result) -> (Color, str):
          This is arguably a bug.
     """
 
-    lsc = builtins.__xonsh__.env["LS_COLORS"]
+    lsc = builtins.__xonsh__.env["LS_COLORS"]  # type:ignore
     color_key = "fi"
 
     # if symlink, get info on (final) target
@@ -1438,7 +1441,7 @@ def color_file(file_path: str, path_stat: os.stat_result) -> (Color, str):
         color_key = "bd"
     elif stat.S_ISCHR(mode):
         color_key = "cd"
-    elif stat.S_ISDOOR(mode):
+    elif stat.S_ISDOOR(mode):  # type:ignore
         color_key = "do"
     else:
         color_key = "or"  # any other type --> orphan
@@ -1624,7 +1627,7 @@ class XonshConsoleLexer(XonshLexer):
 
     name = "Xonsh console lexer"
     aliases = ["xonshcon"]
-    filenames = []
+    filenames: tp.List[str] = []
 
     tokens = {
         "root": [

xonsh/readline_shell.py

@@ -18,7 +18,6 @@ import shutil
 import builtins
 import importlib
 import threading
-import subprocess
 import collections
 
 import xonsh.completers.tools as xct
@@ -32,7 +31,6 @@ from xonsh.ansi_colors import (
 from xonsh.prompt.base import multiline_prompt
 from xonsh.tools import (
     print_exception,
-    check_for_partial_string,
     to_bool,
     columnize,
     carriage_return,
@@ -464,7 +462,7 @@ class ReadlineShell(BaseShell, cmd.Cmd):
             raise ValueError("query completions flag not understood.")
 
     # tab complete on first index too
-    completenames = completedefault
+    completenames = completedefault  # type:ignore
 
     def _load_remaining_input_into_queue(self):
         buf = b""

xonsh/tokenize.py

@@ -78,13 +78,14 @@ from token import (
     VBAREQUAL,
     tok_name,
 )
+import typing as tp
 
 from xonsh.lazyasd import LazyObject
 from xonsh.platform import PYTHON_VERSION_INFO
 
 HAS_WALRUS = PYTHON_VERSION_INFO > (3, 8)
 if HAS_WALRUS:
-    from token import COLONEQUAL
+    from token import COLONEQUAL  # type:ignore
 
 cookie_re = LazyObject(
     lambda: re.compile(r"^[ \t\f]*#.*coding[:=][ \t]*([-\w.]+)", re.ASCII),
@@ -99,7 +100,7 @@ blank_re = LazyObject(
 # token modifications
 #
 tok_name = tok_name.copy()
-__all__ = token.__all__ + [
+__all__ = token.__all__ + [  # type:ignore
     "COMMENT",
     "tokenize",
     "detect_encoding",
@@ -116,11 +117,11 @@ __all__ = token.__all__ + [
 ]
 HAS_ASYNC = PYTHON_VERSION_INFO < (3, 7, 0)
 if HAS_ASYNC:
-    ASYNC = token.ASYNC
-    AWAIT = token.AWAIT
+    ASYNC = token.ASYNC  # type:ignore
+    AWAIT = token.AWAIT  # type:ignore
     ADDSPACE_TOKS = (NAME, NUMBER, ASYNC, AWAIT)
 else:
-    ADDSPACE_TOKS = (NAME, NUMBER)
+    ADDSPACE_TOKS = (NAME, NUMBER)  # type:ignore
 del token  # must clean up token
 
 if HAS_WALRUS:
@@ -128,7 +129,6 @@ if HAS_WALRUS:
 else:
     AUGASSIGN_OPS = r"[+\-*/%&@|^=<>]=?"
 
-
 COMMENT = N_TOKENS
 tok_name[COMMENT] = "COMMENT"
 NL = N_TOKENS + 1
@@ -177,7 +177,7 @@ for v in _xonsh_tokens.values():
     __all__.append(v)
 del _glbs, v
 
-EXACT_TOKEN_TYPES = {
+EXACT_TOKEN_TYPES: tp.Dict[str, tp.Union[str, int]] = {
     "(": LPAR,
     ")": RPAR,
     "[": LSQB,
@@ -330,10 +330,12 @@ _redir_map = (
     "1>2",
 )
 IORedirect = group(group(*_redir_map), "{}>>?".format(group(*_redir_names)))
-_redir_check = set(_redir_map)
-_redir_check = {"{}>".format(i) for i in _redir_names}.union(_redir_check)
-_redir_check = {"{}>>".format(i) for i in _redir_names}.union(_redir_check)
-_redir_check = frozenset(_redir_check)
+
+_redir_check_0 = set(_redir_map)
+_redir_check_1 = {"{}>".format(i) for i in _redir_names}.union(_redir_check_0)
+_redir_check_2 = {"{}>>".format(i) for i in _redir_names}.union(_redir_check_1)
+_redir_check = frozenset(_redir_check_2)
+
 Operator = group(
     r"\*\*=?",
     r">>=?",

xonsh/tools.py

@@ -38,6 +38,7 @@ import warnings
 import operator
 import ast
 import string
+import typing as tp
 
 # adding imports from further xonsh modules is discouraged to avoid circular
 # dependencies
@@ -292,7 +293,7 @@ def FORMATTER():
 class DefaultNotGivenType(object):
     """Singleton for representing when no default value is given."""
 
-    __inst = None
+    __inst: tp.Optional["DefaultNotGivenType"] = None
 
     def __new__(cls):
         if DefaultNotGivenType.__inst is None:
@@ -727,7 +728,7 @@ def fallback(cond, backup):
 # Copyright (c) Python Software Foundation. All rights reserved.
 class _RedirectStream:
 
-    _stream = None
+    _stream: tp.Optional[str] = None
 
     def __init__(self, new_target):
         self._new_target = new_target
@@ -1678,7 +1679,7 @@ _year_to_sec = lambda x: 365.25 * _day_to_sec(x)
 _kb_to_b = lambda x: 1024 * int(x)
 _mb_to_b = lambda x: 1024 * _kb_to_b(x)
 _gb_to_b = lambda x: 1024 * _mb_to_b(x)
-_tb_to_b = lambda x: 1024 * _tb_to_b(x)
+_tb_to_b = lambda x: 1024 * _tb_to_b(x)  # type: ignore
 
 CANON_HISTORY_UNITS = LazyObject(
     lambda: frozenset(["commands", "files", "s", "b"]), globals(), "CANON_HISTORY_UNITS"

xonsh/tracer.py

@@ -8,6 +8,7 @@ import linecache
 import importlib
 import functools
 from inspect import getouterframes
+import typing as tp
 
 from xonsh.lazyasd import LazyObject
 from xonsh.platform import HAS_PYGMENTS
@@ -29,7 +30,7 @@ class TracerType(object):
     state. This is a singleton.
     """
 
-    _inst = None
+    _inst: tp.Optional["TracerType"] = None
     valid_events = frozenset(["line", "call"])
 
     def __new__(cls, *args, **kwargs):

xonsh/webconfig/main.py

@@ -7,9 +7,9 @@ import socketserver
 from http import server
 from pprint import pprint
 from argparse import ArgumentParser
+import typing as tp
 
-
-RENDERERS = []
+RENDERERS: tp.List[tp.Callable] = []
 
 
 def renderer(f):

xonsh/wizard.py

@@ -9,6 +9,7 @@ import fnmatch
 import builtins
 import textwrap
 import collections.abc as cabc
+import typing as tp
 
 from xonsh.tools import to_bool, to_bool_or_break, backup_file, print_color
 from xonsh.jsonutils import serialize_xonsh_json
@@ -20,7 +21,7 @@ from xonsh.jsonutils import serialize_xonsh_json
 class Node(object):
     """Base type of all nodes."""
 
-    attrs = ()
+    attrs: tp.Union[tp.Tuple[str, ...], str] = ()
 
     def __str__(self):
         return PrettyFormatter(self).visit()
@@ -81,7 +82,13 @@ class Question(Node):
 class Input(Node):
     """Gets input from the user."""
 
-    attrs = ("prompt", "converter", "show_conversion", "confirm", "path")
+    attrs: tp.Tuple[str, ...] = (
+        "prompt",
+        "converter",
+        "show_conversion",
+        "confirm",
+        "path",
+    )
 
     def __init__(
         self,
@@ -251,7 +258,7 @@ class StateFile(Input):
     given file name. This node type is likely not useful on its own.
     """
 
-    attrs = ("default_file", "check", "ask_filename")
+    attrs: tp.Tuple[str, ...] = ("default_file", "check", "ask_filename")
 
     def __init__(self, default_file=None, check=True, ask_filename=True):
         """
@@ -604,7 +611,7 @@ class UnstorableType(object):
     singleton.
     """
 
-    _inst = None
+    _inst: tp.Optional["UnstorableType"] = None
 
     def __new__(cls, *args, **kwargs):
         if cls._inst is None:

xonsh/xonfig.py

@@ -15,7 +15,7 @@ import functools
 import itertools
 import contextlib
 import collections
-
+import typing as tp
 
 from xonsh.ply import ply
 
@@ -139,7 +139,7 @@ WIZARD_TAIL = """
 Thanks for using the xonsh configuration wizard!"""
 
 
-_XONFIG_SOURCE_FOREIGN_SHELL_COMMAND = collections.defaultdict(
+_XONFIG_SOURCE_FOREIGN_SHELL_COMMAND: tp.Dict[str, str] = collections.defaultdict(
     lambda: "source-foreign", bash="source-bash", cmd="source-cmd", zsh="source-zsh"
 )
 

xonsh/xoreutils/uptime.py

@@ -15,13 +15,13 @@ import sys
 import time
 import ctypes
 import struct
+import typing as tp
 
 import xonsh.platform as xp
 import xonsh.lazyimps as xlimps
 import xonsh.lazyasd as xl
 
-
-_BOOTTIME = None
+_BOOTTIME: tp.Optional[float] = None
 
 
 def _uptime_osx():

xontrib/voxapi.py

@@ -18,7 +18,7 @@ import subprocess as sp
 
 
 from xonsh.platform import ON_POSIX, ON_WINDOWS
-from xonsh.fs import PathLike, fspath
+
 
 # This is because builtins aren't globally created during testing.
 # FIXME: Is there a better way?
@@ -160,8 +160,8 @@ class Vox(collections.abc.Mapping):
         )
         # NOTE: clear=True is the same as delete then create.
         # NOTE: upgrade=True is its own method
-        if isinstance(name, PathLike):
-            env_path = fspath(name)
+        if isinstance(name, os.PathLike):
+            env_path = os.fspath(name)
         else:
             env_path = os.path.join(self.venvdir, name)
         if not self._check_reserved(env_path):
@@ -274,8 +274,8 @@ class Vox(collections.abc.Mapping):
         if name is ...:
             env = builtins.__xonsh__.env
             env_paths = [env["VIRTUAL_ENV"]]
-        elif isinstance(name, PathLike):
-            env_paths = [fspath(name)]
+        elif isinstance(name, os.PathLike):
+            env_paths = [os.fspath(name)]
         else:
             if not self._check_reserved(name):
                 # Don't allow a venv that could be a venv special dir