Add more type annotations and improve MonkeyType tooling

mypy.ini

@@ -1,13 +1,32 @@
 [mypy]
-; TODO: check_untyped_defs = True
 ; TODO: Make this more granular; docs recommend it as a "last resort"
-; https://mypy.readthedocs.io/en/latest/existing_code.html#start-small
+; https://mypy.readthedocs.io/en/latest/running_mypy.html#missing-type-hints-for-third-party-library
 ignore_missing_imports = True
-; NOTE: Docs recommend `normal` or `silent` for an existing codebase
-; https://mypy.readthedocs.io/en/latest/running_mypy.html#following-imports
-; follow_imports = skip
+
+warn_redundant_casts = True
+warn_unused_configs = True
+warn_unused_ignores = True
+
+; Reporting
 show_traceback = True
 html_report = mypy
 linecount_report = mypy
 lineprecision_report = mypy
 txt_report = mypy
+
+; TODO: Adopt --strict settings, iterating towards something like:
+; https://github.com/pypa/packaging/blob/master/setup.cfg
+; Starting with modules that have annotations applied via MonkeyType
+[mypy-twine.auth,twine.cli,twine.package,twine.repository,twine.utils]
+; Enabling this will fail on subclasses of untype imports, e.g. tqdm
+; disallow_subclassing_any = True
+disallow_any_generics = True
+disallow_untyped_calls = True
+disallow_untyped_defs = True
+disallow_incomplete_defs = True
+check_untyped_defs = True
+disallow_untyped_decorators = True
+no_implicit_optional = True
+warn_return_any = True
+no_implicit_reexport = True
+strict_equality = True

tox.ini

@@ -51,6 +51,14 @@ commands =
     # TODO: Consider checking the tests after the source is fully typed
     mypy twine
 
+[testenv:monkeytype]
+deps = 
+    {[testenv]deps}
+    {[testenv:typing]deps}
+    monkeytype
+commands = 
+    monkeytype {posargs:run -m pytest}
+
 [testenv:lint]
 deps =
     {[testenv:code-style]deps}

twine/auth.py

@@ -3,6 +3,8 @@
 import warnings
 from typing import Callable
 from typing import Optional
+from typing import Type
+from typing import cast
 
 import keyring
 
@@ -11,18 +13,18 @@
 
 
 class CredentialInput:
-    def __init__(self, username: str = None, password: str = None):
+    def __init__(self, username: str = None, password: str = None) -> None:
         self.username = username
         self.password = password
 
 
 class Resolver:
-    def __init__(self, config: utils.RepositoryConfig, input: CredentialInput):
+    def __init__(self, config: utils.RepositoryConfig, input: CredentialInput) -> None:
         self.config = config
         self.input = input
 
     @classmethod
-    def choose(cls, interactive):
+    def choose(cls, interactive: bool) -> Type["Resolver"]:
         return cls if interactive else Private
 
     @property  # type: ignore  # https://github.com/python/mypy/issues/1362
@@ -53,7 +55,7 @@ def get_username_from_keyring(self) -> Optional[str]:
         try:
             creds = keyring.get_credential(self.system, None)
             if creds:
-                return creds.username
+                return cast(str, creds.username)
         except AttributeError:
             # To support keyring prior to 15.2
             pass
@@ -63,7 +65,7 @@ def get_username_from_keyring(self) -> Optional[str]:
 
     def get_password_from_keyring(self) -> Optional[str]:
         try:
-            return keyring.get_password(self.system, self.username)
+            return cast(str, keyring.get_password(self.system, self.username))
         except Exception as exc:
             warnings.warn(str(exc))
         return None
@@ -76,10 +78,10 @@ def password_from_keyring_or_prompt(self) -> str:
             "password", getpass.getpass
         )
 
-    def prompt(self, what: str, how: Callable) -> str:
+    def prompt(self, what: str, how: Callable[..., str]) -> str:
         return how(f"Enter your {what}: ")
 
 
 class Private(Resolver):
-    def prompt(self, what: str, how: Optional[Callable] = None) -> str:
+    def prompt(self, what: str, how: Optional[Callable[..., str]] = None) -> str:
         raise exceptions.NonInteractive(f"Credential not found for {what}.")

twine/cli.py

@@ -12,6 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import argparse
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Tuple
 
 import pkg_resources
 import pkginfo
@@ -24,12 +28,14 @@
 from twine import _installed
 
 
-def _registered_commands(group="twine.registered_commands"):
+def _registered_commands(
+    group: str = "twine.registered_commands",
+) -> Dict[str, pkg_resources.EntryPoint]:
     registered_commands = pkg_resources.iter_entry_points(group=group)
     return {c.name: c for c in registered_commands}
 
 
-def list_dependencies_and_versions():
+def list_dependencies_and_versions() -> List[Tuple[str, str]]:
     return [
         ("pkginfo", _installed.Installed(pkginfo).version),
         ("requests", requests.__version__),
@@ -39,13 +45,13 @@ def list_dependencies_and_versions():
     ]
 
 
-def dep_versions():
+def dep_versions() -> str:
     return ", ".join(
         "{}: {}".format(*dependency) for dependency in list_dependencies_and_versions()
     )
 
 
-def dispatch(argv):
+def dispatch(argv: List[str]) -> Any:
     registered_commands = _registered_commands()
     parser = argparse.ArgumentParser(prog="twine")
     parser.add_argument(

twine/package.py

@@ -159,14 +159,16 @@ def metadata_dictionary(self) -> Dict[str, MetadataValue]:
 
         return data
 
-    def add_gpg_signature(self, signature_filepath: str, signature_filename: str):
+    def add_gpg_signature(
+        self, signature_filepath: str, signature_filename: str
+    ) -> None:
         if self.gpg_signature is not None:
             raise exceptions.InvalidDistribution("GPG Signature can only be added once")
 
         with open(signature_filepath, "rb") as gpg:
             self.gpg_signature = (signature_filename, gpg.read())
 
-    def sign(self, sign_with: str, identity: Optional[str]):
+    def sign(self, sign_with: str, identity: Optional[str]) -> None:
         print(f"Signing {self.basefilename}")
         gpg_args: Tuple[str, ...] = (sign_with, "--detach-sign")
         if identity:
@@ -177,7 +179,7 @@ def sign(self, sign_with: str, identity: Optional[str]):
         self.add_gpg_signature(self.signed_filename, self.signed_basefilename)
 
     @classmethod
-    def run_gpg(cls, gpg_args):
+    def run_gpg(cls, gpg_args: Tuple[str, ...]) -> None:
         try:
             subprocess.check_call(gpg_args)
             return

twine/repository.py

@@ -12,10 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import sys
+from typing import Any
 from typing import Dict
 from typing import List
 from typing import Optional
+from typing import Set
 from typing import Tuple
+from typing import cast
 
 import requests
 import requests_toolbelt
@@ -38,7 +41,7 @@
 
 
 class ProgressBar(tqdm.tqdm):
-    def update_to(self, n):
+    def update_to(self, n: int) -> None:
         """Update the bar in the way compatible with requests-toolbelt.
 
         This is identical to tqdm.update, except ``n`` will be the current
@@ -68,7 +71,8 @@ def __init__(
         for scheme in ("http://", "https://"):
             self.session.mount(scheme, self._make_adapter_with_retries())
 
-        self._releases_json_data: Dict[str, Dict] = {}
+        # Working around https://github.com/python/typing/issues/182
+        self._releases_json_data: Dict[str, Dict[str, Any]] = {}
         self.disable_progress_bar = disable_progress_bar
 
     @staticmethod
@@ -86,14 +90,16 @@ def _make_user_agent_string() -> str:
         from twine import cli
 
         dependencies = cli.list_dependencies_and_versions()
-        return (
-            user_agent.UserAgentBuilder("twine", twine.__version__,)
+        user_agent_string = (
+            user_agent.UserAgentBuilder("twine", twine.__version__)
             .include_extras(dependencies)
             .include_implementation()
             .build()
         )
 
-    def close(self):
+        return cast(str, user_agent_string)
+
+    def close(self) -> None:
         self.session.close()
 
     @staticmethod
@@ -117,7 +123,7 @@ def set_client_certificate(self, clientcert: Optional[str]) -> None:
         if clientcert:
             self.session.cert = clientcert
 
-    def register(self, package):
+    def register(self, package: package_file.PackageFile) -> requests.Response:
         data = package.metadata_dictionary()
         data.update({":action": "submit", "protocol_version": "1"})
 
@@ -232,7 +238,7 @@ def package_is_uploaded(
 
         return False
 
-    def release_urls(self, packages):
+    def release_urls(self, packages: List[package_file.PackageFile]) -> Set[str]:
         if self.url.startswith(WAREHOUSE):
             url = WAREHOUSE_WEB
         elif self.url.startswith(TEST_WAREHOUSE):
@@ -245,7 +251,7 @@ def release_urls(self, packages):
             for package in packages
         }
 
-    def verify_package_integrity(self, package: package_file.PackageFile):
+    def verify_package_integrity(self, package: package_file.PackageFile) -> None:
         # TODO(sigmavirus24): Add a way for users to download the package and
         # check it's hash against what it has locally.
         pass

twine/settings.py

@@ -135,12 +135,17 @@ def __init__(
         )
 
     @property
-    def username(self):
-        return self.auth.username
+    def username(self) -> Optional[str]:
+        # Workaround for https://github.com/python/mypy/issues/5858
+        return cast(Optional[str], self.auth.username)
 
     @property
-    def password(self):
-        return None if self.client_cert else self.auth.password
+    def password(self) -> Optional[str]:
+        if self.client_cert:
+            return None
+
+        # Workaround for https://github.com/python/mypy/issues/5858
+        return cast(Optional[str], self.auth.password)
 
     @staticmethod
     def register_argparse_arguments(parser: argparse.ArgumentParser) -> None:

twine/utils.py

@@ -17,10 +17,13 @@
 import functools
 import os
 import os.path
+from typing import Any
 from typing import Callable
 from typing import DefaultDict
 from typing import Dict
 from typing import Optional
+from typing import Sequence
+from typing import Union
 from urllib.parse import urlparse
 from urllib.parse import urlunparse
 
@@ -181,7 +184,7 @@ def get_userpass_value(
     cli_value: Optional[str],
     config: RepositoryConfig,
     key: str,
-    prompt_strategy: Optional[Callable] = None,
+    prompt_strategy: Optional[Callable[[], str]] = None,
 ) -> Optional[str]:
     """Gets the username / password from config.
 
@@ -221,33 +224,49 @@ class EnvironmentDefault(argparse.Action):
     """Get values from environment variable."""
 
     def __init__(
-        self, env: str, required: bool = True, default: Optional[str] = None, **kwargs
+        self,
+        env: str,
+        required: bool = True,
+        default: Optional[str] = None,
+        **kwargs: Any,
     ) -> None:
         default = os.environ.get(env, default)
         self.env = env
         if default:
             required = False
         super().__init__(default=default, required=required, **kwargs)
 
-    def __call__(self, parser, namespace, values, option_string=None):
+    def __call__(
+        self,
+        parser: argparse.ArgumentParser,
+        namespace: argparse.Namespace,
+        values: Union[str, Sequence[Any], None],
+        option_string: Optional[str] = None,
+    ) -> None:
         setattr(namespace, self.dest, values)
 
 
 class EnvironmentFlag(argparse.Action):
     """Set boolean flag from environment variable."""
 
-    def __init__(self, env: str, **kwargs) -> None:
+    def __init__(self, env: str, **kwargs: Any) -> None:
         default = self.bool_from_env(os.environ.get(env))
         self.env = env
         super().__init__(default=default, nargs=0, **kwargs)
 
-    def __call__(self, parser, namespace, values, option_string=None):
+    def __call__(
+        self,
+        parser: argparse.ArgumentParser,
+        namespace: argparse.Namespace,
+        values: Union[str, Sequence[Any], None],
+        option_string: Optional[str] = None,
+    ) -> None:
         setattr(namespace, self.dest, True)
 
     @staticmethod
-    def bool_from_env(val):
+    def bool_from_env(val: Optional[str]) -> bool:
         """
         Allow '0' and 'false' and 'no' to be False
         """
         falsey = {"0", "false", "no"}
-        return val and val.lower() not in falsey
+        return bool(val and val.lower() not in falsey)