add type annotations to support static type checks

Author: chrissimpkins

lib/fdiff/__main__.py

@@ -3,15 +3,16 @@
 import os
 import sys
 import argparse
+from typing import Iterator, Iterable, List, Optional, Text, Tuple
 
-from fdiff import __version__
-from fdiff.color import color_unified_diff_line
-from fdiff.diff import external_diff, u_diff
-from fdiff.textiter import head, tail
-from fdiff.utils import file_exists, get_tables_argument_list
+from . import __version__
+from .color import color_unified_diff_line
+from .diff import external_diff, u_diff
+from .textiter import head, tail
+from .utils import file_exists, get_tables_argument_list
 
 
-def main():  # pragma: no cover
+def main() -> None:  # pragma: no cover
     # try/except block rationale:
     # handles "premature" socket closure exception that is
     # raised by Python when stdout is piped to tools like
@@ -27,13 +28,11 @@ def main():  # pragma: no cover
         sys.exit(0)
 
 
-def run(argv):
+def run(argv: List[Text]) -> None:
     # ------------------------------------------
     # argparse command line argument definitions
     # ------------------------------------------
-    parser = argparse.ArgumentParser(
-        description="An OpenType table diff tool for fonts."
-    )
+    parser = argparse.ArgumentParser(description="An OpenType table diff tool for fonts.")
     parser.add_argument("--version", action="version", version=f"fdiff v{__version__}")
     parser.add_argument(
         "-c",
@@ -66,7 +65,7 @@ def run(argv):
     parser.add_argument("PREFILE", help="Font file path/URL 1")
     parser.add_argument("POSTFILE", help="Font file path/URL 2")
 
-    args = parser.parse_args(argv)
+    args: argparse.Namespace = parser.parse_args(argv)
 
     # /////////////////////////////////////////////////////////
     #
@@ -110,12 +109,12 @@ def run(argv):
     # the command line arguments
     # set as a Python list if it was defined on the command line
     # or as None if it was not set on the command line
-    include_list = get_tables_argument_list(args.include)
-    exclude_list = get_tables_argument_list(args.exclude)
+    include_list: Optional[List[Text]] = get_tables_argument_list(args.include)
+    exclude_list: Optional[List[Text]] = get_tables_argument_list(args.exclude)
 
     # flip logic of the command line flag for multi process
     # optimization use
-    use_mp = not args.nomp
+    use_mp: bool = not args.nomp
 
     if args.external:
         # ------------------------------
@@ -138,7 +137,7 @@ def run(argv):
             sys.exit(1)
 
         try:
-            diff = external_diff(
+            ext_diff: Iterable[Tuple[Text, Optional[int]]] = external_diff(
                 args.external,
                 args.PREFILE,
                 args.POSTFILE,
@@ -148,7 +147,7 @@ def run(argv):
             )
 
             # write stdout from external tool
-            for line, exit_code in diff:
+            for line, exit_code in ext_diff:
                 # format with color if color flag is entered on command line
                 if args.color:
                     sys.stdout.write(color_unified_diff_line(line))
@@ -165,7 +164,7 @@ def run(argv):
         # ---------------
         # perform the unified diff analysis
         try:
-            diff = u_diff(
+            uni_diff: Iterator[Text] = u_diff(
                 args.PREFILE,
                 args.POSTFILE,
                 context_lines=args.lines,
@@ -180,11 +179,11 @@ def run(argv):
         # re-define the line contents of the diff iterable
         # if head or tail is requested
         if args.head:
-            iterable = head(diff, args.head)
+            iterable = head(uni_diff, args.head)
         elif args.tail:
-            iterable = tail(diff, args.tail)
+            iterable = tail(uni_diff, args.tail)
         else:
-            iterable = diff
+            iterable = uni_diff
 
         # print unified diff results to standard output stream
         has_diff = False

lib/fdiff/aio.py

@@ -1,9 +1,14 @@
 #!/usr/bin/env python3
 
-import aiofiles
+import os
+from typing import AnyStr, Text, Union
 
+import aiofiles  # type: ignore
 
-async def async_write_bin(path, binary):
+
+async def async_write_bin(
+    path: Union[AnyStr, "os.PathLike[Text]"], binary: bytes
+) -> None:
     """Asynchronous IO writes of binary data `binary` to disk on the file path `path`"""
-    async with aiofiles.open(path, "wb") as f:
+    async with aiofiles.open(path, "wb") as f:  # type: ignore
         await f.write(binary)

lib/fdiff/color.py

@@ -1,6 +1,8 @@
 #!/usr/bin/env python3
 
-ansicolors = {
+from typing import Dict, Text
+
+ansicolors: Dict[Text, Text] = {
     "BLACK": "\033[30m",
     "RED": "\033[31m",
     "GREEN": "\033[32m",
@@ -13,13 +15,13 @@
     "RESET": "\033[0m",
 }
 
-green_start = ansicolors["GREEN"]
-red_start = ansicolors["RED"]
-cyan_start = ansicolors["CYAN"]
-reset = ansicolors["RESET"]
+green_start: Text = ansicolors["GREEN"]
+red_start: Text = ansicolors["RED"]
+cyan_start: Text = ansicolors["CYAN"]
+reset: Text = ansicolors["RESET"]
 
 
-def color_unified_diff_line(line):
+def color_unified_diff_line(line: Text) -> Text:
     """Returns an ANSI escape code colored string with color based
     on the unified diff line type."""
     if line[0:2] == "+ ":

lib/fdiff/diff.py

@@ -7,16 +7,17 @@
 import shlex
 import subprocess
 import tempfile
+from typing import Any, Iterator, Iterable, List, Optional, Text, Tuple
 
-from fontTools.ttLib import TTFont
+from fontTools.ttLib import TTFont  # type: ignore
 
-from fdiff.exceptions import AIOError
-from fdiff.remote import (
+from .exceptions import AIOError
+from .remote import (
     _get_filepath_from_url,
     create_async_get_request_session_and_run,
 )
 
-from fdiff.utils import get_file_modtime
+from .utils import get_file_modtime
 
 
 #
@@ -26,7 +27,7 @@
 #
 
 
-def _async_fetch_files(dirpath, urls):
+def _async_fetch_files(dirpath: Text, urls: List[Text]) -> None:
     loop = asyncio.get_event_loop()
     tasks = loop.run_until_complete(
         create_async_get_request_session_and_run(urls, dirpath)
@@ -45,8 +46,13 @@ def _async_fetch_files(dirpath, urls):
 
 
 def _get_fonts_and_save_xml(
-    filepath_a, filepath_b, tmpdirpath, include_tables, exclude_tables, use_multiprocess
-):
+    filepath_a: Text,
+    filepath_b: Text,
+    tmpdirpath: Text,
+    include_tables: Optional[List[Text]],
+    exclude_tables: Optional[List[Text]],
+    use_multiprocess: bool,
+) -> Tuple[Text, Text, Text, Text, Text, Text]:
     post_pathname, postpath, pre_pathname, prepath = _get_pre_post_paths(
         filepath_a, filepath_b, tmpdirpath
     )
@@ -69,8 +75,12 @@ def _get_fonts_and_save_xml(
     return left_ttxpath, right_ttxpath, pre_pathname, prepath, post_pathname, postpath
 
 
-def _get_pre_post_paths(filepath_a, filepath_b, dirpath):
-    urls = []
+def _get_pre_post_paths(
+    filepath_a: Text,
+    filepath_b: Text,
+    dirpath: Text,
+) -> Tuple[Text, Text, Text, Text]:
+    urls: List[Text] = []
     if filepath_a.startswith("http"):
         urls.append(filepath_a)
         prepath = _get_filepath_from_url(filepath_a, dirpath)
@@ -94,14 +104,14 @@ def _get_pre_post_paths(filepath_a, filepath_b, dirpath):
 
 
 def _mp_save_ttx_xml(
-    tt_left,
-    tt_right,
-    left_ttxpath,
-    right_ttxpath,
-    exclude_tables,
-    include_tables,
-    use_multiprocess,
-):
+    tt_left: Any,
+    tt_right: Any,
+    left_ttxpath: Text,
+    right_ttxpath: Text,
+    exclude_tables: Optional[List[Text]],
+    include_tables: Optional[List[Text]],
+    use_multiprocess: bool,
+) -> None:
     if use_multiprocess and cpu_count() > 1:
         # Use parallel fontTools.ttLib.TTFont.saveXML dump
         # by default on multi CPU systems.  This is a performance
@@ -121,13 +131,20 @@ def _mp_save_ttx_xml(
         _ttfont_save_xml(tt_right, right_ttxpath, include_tables, exclude_tables)
 
 
-def _ttfont_save_xml(ttf, filepath, include_tables, exclude_tables):
+def _ttfont_save_xml(
+    ttf: Any,
+    filepath: Text,
+    include_tables: Optional[List[Text]],
+    exclude_tables: Optional[List[Text]],
+) -> bool:
     """Writes TTX specification formatted XML to disk on filepath."""
     ttf.saveXML(filepath, tables=include_tables, skipTables=exclude_tables)
     return True
 
 
-def _validate_table_excludes(exclude_tables, tt_left, tt_right):
+def _validate_table_excludes(
+    exclude_tables: Optional[List[Text]], tt_left: Any, tt_right: Any
+) -> None:
     # Validation: exclude_tables request should be for tables that are in one of
     # the two fonts.  Mis-specified OT table definitions could otherwise result
     # in the presence of a table in the diff when the request was to exclude it.
@@ -140,7 +157,9 @@ def _validate_table_excludes(exclude_tables, tt_left, tt_right):
                 )
 
 
-def _validate_table_includes(include_tables, tt_left, tt_right):
+def _validate_table_includes(
+    include_tables: Optional[List[Text]], tt_left: Any, tt_right: Any
+) -> None:
     # Validation: include_tables request should be for tables that are in one of
     # the two fonts. This otherwise silently passes with exit status code 0 which
     # could lead to the interpretation of no diff between two files when the table
@@ -164,13 +183,13 @@ def _validate_table_includes(include_tables, tt_left, tt_right):
 
 
 def u_diff(
-    filepath_a,
-    filepath_b,
-    context_lines=3,
-    include_tables=None,
-    exclude_tables=None,
-    use_multiprocess=True,
-):
+    filepath_a: Text,
+    filepath_b: Text,
+    context_lines: int = 3,
+    include_tables: Optional[List[Text]] = None,
+    exclude_tables: Optional[List[Text]] = None,
+    use_multiprocess: bool = True,
+) -> Iterator[Text]:
     """Performs a unified diff on a TTX serialized data format dump of font binary data using
     a modified version of the Python standard libary difflib module.
 
@@ -230,13 +249,13 @@ def u_diff(
 
 
 def external_diff(
-    command,
-    filepath_a,
-    filepath_b,
-    include_tables=None,
-    exclude_tables=None,
-    use_multiprocess=True,
-):
+    command: Text,
+    filepath_a: Text,
+    filepath_b: Text,
+    include_tables: Optional[List[Text]] = None,
+    exclude_tables: Optional[List[Text]] = None,
+    use_multiprocess: bool = True,
+) -> Iterable[Tuple[Text, Optional[int]]]:
     """Performs a unified diff on a TTX serialized data format dump of font binary data using
     an external diff executable that is requested by the caller via `command`
 
@@ -285,10 +304,10 @@ def external_diff(
         )
 
         while True:
-            output = process.stdout.readline()
+            output = process.stdout.readline()  # type: ignore
             exit_status = process.poll()
             if len(output) == 0 and exit_status is not None:
-                err = process.stderr.read()
+                err = process.stderr.read()  # type: ignore
                 if err:
                     raise IOError(err)
                 yield output, exit_status

lib/fdiff/remote.py

@@ -3,23 +3,31 @@
 import os.path
 import urllib.parse
 
-from collections import namedtuple
+# from collections import namedtuple
+from typing import Any, AnyStr, List, NamedTuple, Optional, Text, Tuple
 
-import aiohttp
+import aiohttp  # type: ignore
 import asyncio
 
-from fdiff.aio import async_write_bin
+from .aio import async_write_bin
 
 
-def _get_filepath_from_url(url, dirpath):
+class FWRes(NamedTuple):
+    url: Text
+    filepath: Optional[Text]
+    http_status: int
+    write_success: bool
+
+
+def _get_filepath_from_url(url: Text, dirpath: Text) -> Text:
     """Returns filepath from base file name in URL and directory path."""
     url_path_list = urllib.parse.urlsplit(url)
     abs_filepath = url_path_list.path
     basepath = os.path.split(abs_filepath)[-1]
     return os.path.join(dirpath, basepath)
 
 
-async def async_fetch(session, url):
+async def async_fetch(session: Any, url: AnyStr) -> Tuple[AnyStr, Any, Any]:
     """Asynchronous I/O HTTP GET request with a ClientSession instantiated
     from the aiohttp library."""
     async with session.get(url) as response:
@@ -31,15 +39,12 @@ async def async_fetch(session, url):
         return url, status, binary
 
 
-async def async_fetch_and_write(session, url, dirpath):
+async def async_fetch_and_write(session: Any, url: Text, dirpath: Text) -> FWRes:
     """Asynchronous I/O HTTP GET request with a ClientSession instantiated
     from the aiohttp library, followed by an asynchronous I/O file write of
     the binary to disk with the aiofiles library.
 
     :returns `FWRes` namedtuple with url, filepath, http_status, write_success fields"""
-    FWResponse = namedtuple(
-        "FWRes", ["url", "filepath", "http_status", "write_success"]
-    )
     url, status, binary = await async_fetch(session, url)
     if status != 200:
         filepath = None
@@ -49,12 +54,14 @@ async def async_fetch_and_write(session, url, dirpath):
         await async_write_bin(filepath, binary)
         write_success = True
 
-    return FWResponse(
+    return FWRes(
         url=url, filepath=filepath, http_status=status, write_success=write_success
     )
 
 
-async def create_async_get_request_session_and_run(urls, dirpath):
+async def create_async_get_request_session_and_run(
+    urls: List[Text], dirpath: Text
+) -> List[Any]:
     """Creates an aiohttp library ClientSession and performs asynchronous GET requests +
     binary file writes with the binary response from the GET request.