Add type hints and run mypy

.github/workflows/linters.yml

@@ -59,3 +59,22 @@ jobs:
       - uses: actions/checkout@v2
       - uses: actions/setup-python@v2
       - uses: psf/black@main
+
+  mypy:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python 3.9
+      uses: actions/setup-python@v2
+      with:
+        python-version: 3.9
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements-lint.txt
+    - name: Run mypy
+      run: |
+        mypy port_for/ tests/ scripts/port-for

mypy.ini

@@ -0,0 +1,25 @@
+[mypy]
+allow_redefinition = False
+allow_untyped_globals = False
+check_untyped_defs = True
+disallow_incomplete_defs = True
+disallow_subclassing_any = True
+disallow_untyped_calls = True
+disallow_untyped_decorators = True
+disallow_untyped_defs = True
+follow_imports = silent
+ignore_missing_imports = False
+implicit_reexport = False
+no_implicit_optional = True
+pretty = True
+show_error_codes = True
+strict_equality = True
+warn_no_return = True
+warn_return_any = True
+warn_unreachable = True
+warn_unused_ignores = True
+
+# Bundled third-party package.
+[mypy-port_for.docopt.*]
+check_untyped_defs = False
+disallow_untyped_defs = False

port_for/__init__.py

@@ -3,6 +3,7 @@
 
 __version__ = "0.5.0"
 
+from ._ranges import UNASSIGNED_RANGES
 from .api import (
     available_good_ports,
     available_ports,
@@ -11,7 +12,19 @@
     port_is_used,
     select_random,
     get_port,
-    UNASSIGNED_RANGES,
 )
 from .store import PortStore
 from .exceptions import PortForException
+
+__all__ = (
+    "UNASSIGNED_RANGES",
+    "available_good_ports",
+    "available_ports",
+    "is_available",
+    "good_port_ranges",
+    "port_is_used",
+    "select_random",
+    "get_port",
+    "PortStore",
+    "PortForException",
+)

port_for/_download_ranges.py

@@ -9,6 +9,7 @@
 import datetime
 from urllib.request import Request, urlopen
 from xml.etree import ElementTree
+from typing import Set, Iterator, Iterable, Tuple
 
 from port_for.utils import to_ranges, ranges_to_set
 
@@ -25,7 +26,7 @@
 WIKIPEDIA_PAGE = "http://en.wikipedia.org/wiki/List_of_TCP_and_UDP_port_numbers"
 
 
-def _write_unassigned_ranges(out_filename):
+def _write_unassigned_ranges(out_filename: str) -> None:
     """
     Downloads ports data from IANA & Wikipedia and converts
     it to a python module. This function is used to generate _ranges.py.
@@ -41,14 +42,14 @@ def _write_unassigned_ranges(out_filename):
         f.write("]\n")
 
 
-def _unassigned_ports():
+def _unassigned_ports() -> Set[int]:
     """Return a set of all unassigned ports (according to IANA and Wikipedia)"""
     free_ports = ranges_to_set(_parse_ranges(_iana_unassigned_port_ranges()))
     known_ports = ranges_to_set(_wikipedia_known_port_ranges())
     return free_ports.difference(known_ports)
 
 
-def _wikipedia_known_port_ranges():
+def _wikipedia_known_port_ranges() -> Iterator[Tuple[int, int]]:
     """
     Returns used port ranges according to Wikipedia page.
     This page contains unofficial well-known ports.
@@ -61,21 +62,26 @@ def _wikipedia_known_port_ranges():
     return ((int(p[1]), int(p[3] if p[3] else p[1])) for p in ports)
 
 
-def _iana_unassigned_port_ranges():
+def _iana_unassigned_port_ranges() -> Iterator[str]:
     """
     Returns unassigned port ranges according to IANA.
     """
     page = urlopen(IANA_DOWNLOAD_URL).read()
     xml = ElementTree.fromstring(page)
     records = xml.findall("{%s}record" % IANA_NS)
     for record in records:
-        description = record.find("{%s}description" % IANA_NS).text
+        description_el = record.find("{%s}description" % IANA_NS)
+        assert description_el is not None
+        description = description_el.text
         if description == "Unassigned":
-            numbers = record.find("{%s}number" % IANA_NS).text
+            number_el = record.find("{%s}number" % IANA_NS)
+            assert number_el is not None
+            numbers = number_el.text
+            assert numbers is not None
             yield numbers
 
 
-def _parse_ranges(ranges):
+def _parse_ranges(ranges: Iterable[str]) -> Iterator[Tuple[int, int]]:
     """Converts a list of string ranges to a list of [low, high] tuples."""
     for txt in ranges:
         if "-" in txt:

port_for/api.py

@@ -1,11 +1,11 @@
 # -*- coding: utf-8 -*-
-# -*- coding: utf-8 -*-
 from __future__ import absolute_import, division, with_statement
 import contextlib
 import socket
 import errno
 import random
 from itertools import chain
+from typing import Optional, Set, List, Tuple, Iterable, TypeVar, Type, Union
 from port_for import ephemeral, utils
 from ._ranges import UNASSIGNED_RANGES
 from .exceptions import PortForException
@@ -14,7 +14,10 @@
 SYSTEM_PORT_RANGE = (0, 1024)
 
 
-def select_random(ports=None, exclude_ports=None):
+def select_random(
+    ports: Optional[Set[int]] = None,
+    exclude_ports: Optional[Iterable[int]] = None,
+) -> int:
     """
     Returns random unused port number.
     """
@@ -32,14 +35,18 @@ def select_random(ports=None, exclude_ports=None):
     raise PortForException("Can't select a port")
 
 
-def is_available(port):
+def is_available(port: int) -> bool:
     """
     Returns if port is good to choose.
     """
     return port in available_ports() and not port_is_used(port)
 
 
-def available_ports(low=1024, high=65535, exclude_ranges=None):
+def available_ports(
+    low: int = 1024,
+    high: int = 65535,
+    exclude_ranges: Optional[List[Tuple[int, int]]] = None,
+) -> Set[int]:
     """
     Returns a set of possible ports (excluding system,
     ephemeral and well-known ports).
@@ -56,7 +63,9 @@ def available_ports(low=1024, high=65535, exclude_ranges=None):
     return available.difference(exclude)
 
 
-def good_port_ranges(ports=None, min_range_len=20, border=3):
+def good_port_ranges(
+    ports: Optional[Set[int]] = None, min_range_len: int = 20, border: int = 3
+) -> List[Tuple[int, int]]:
     """
     Returns a list of 'good' port ranges.
     Such ranges are large and don't contain ephemeral or well-known ports.
@@ -76,13 +85,13 @@ def good_port_ranges(ports=None, min_range_len=20, border=3):
     return without_borders
 
 
-def available_good_ports(min_range_len=20, border=3):
+def available_good_ports(min_range_len: int = 20, border: int = 3) -> Set[int]:
     return utils.ranges_to_set(
         good_port_ranges(min_range_len=min_range_len, border=border)
     )
 
 
-def port_is_used(port, host="127.0.0.1"):
+def port_is_used(port: int, host: str = "127.0.0.1") -> bool:
     """
     Returns if port is used. Port is considered used if the current process
     can't bind to it or the port doesn't refuse connections.
@@ -91,7 +100,7 @@ def port_is_used(port, host="127.0.0.1"):
     return not unused
 
 
-def _can_bind(port, host):
+def _can_bind(port: int, host: str) -> bool:
     sock = socket.socket()
     with contextlib.closing(sock):
         try:
@@ -101,20 +110,36 @@ def _can_bind(port, host):
     return True
 
 
-def _refuses_connection(port, host):
+def _refuses_connection(port: int, host: str) -> bool:
     sock = socket.socket()
     with contextlib.closing(sock):
         sock.settimeout(1)
         err = sock.connect_ex((host, port))
         return err == errno.ECONNREFUSED
 
 
-def filter_by_type(lst, type_of):
+T = TypeVar("T")
+
+
+def filter_by_type(lst: Iterable, type_of: Type[T]) -> List[T]:
     """Returns a list of elements with given type."""
     return [e for e in lst if isinstance(e, type_of)]
 
 
-def get_port(ports):
+def get_port(
+    ports: Union[
+        str,
+        int,
+        Tuple[int, int],
+        Set[int],
+        List[str],
+        List[int],
+        List[Tuple[int, int]],
+        List[Set[int]],
+        List[Union[Set[int], Tuple[int, int]]],
+        List[Union[str, int, Tuple[int, int], Set[int]]],
+    ]
+) -> Optional[int]:
     """
     Retuns a random available port. If there's only one port passed
     (e.g. 5000 or '5000') function does not check if port is available.
@@ -125,7 +150,7 @@ def get_port(ports):
         randomly selected port (None) - any random available port
         [(2000,3000)] or (2000,3000) - random available port from a given range
         [{4002,4003}] or {4002,4003} - random of 4002 or 4003 ports
-        [(2000,3000), {4002,4003}] -random of given orange and set
+        [(2000,3000), {4002,4003}] -random of given range and set
     :returns: a random free port
     :raises: ValueError
     """
@@ -135,18 +160,26 @@ def get_port(ports):
         return select_random(None)
 
     try:
-        return int(ports)
+        return int(ports)  # type: ignore[arg-type]
     except TypeError:
         pass
 
-    ports_set = set()
+    ports_set: Set[int] = set()
 
     try:
         if not isinstance(ports, list):
             ports = [ports]
-        ranges = utils.ranges_to_set(filter_by_type(ports, tuple))
-        nums = set(filter_by_type(ports, int))
-        sets = set(chain(*filter_by_type(ports, (set, frozenset))))
+        ranges: Set[int] = utils.ranges_to_set(
+            filter_by_type(ports, tuple)  # type: ignore[arg-type]
+        )
+        nums: Set[int] = set(filter_by_type(ports, int))
+        sets: Set[int] = set(
+            chain(
+                *filter_by_type(
+                    ports, (set, frozenset)  # type: ignore[arg-type]
+                )
+            )
+        )
         ports_set = ports_set.union(ranges, sets, nums)
     except ValueError:
         raise PortForException(
@@ -155,7 +188,7 @@ def get_port(ports):
             'or "(4000,5000)" or a comma-separated ports set'
             '"[{4000,5000,6000}]" or list of ints "[400,5000,6000,8000]"'
             'or all of them "[(20000, 30000), {48889, 50121}, 4000, 4004]"'
-            % ports
+            % (ports,)
         )
 
     return select_random(ports_set)

port_for/ephemeral.py

@@ -8,11 +8,12 @@
 """
 from __future__ import absolute_import, with_statement
 import subprocess
+from typing import List, Tuple, Dict
 
 DEFAULT_EPHEMERAL_PORT_RANGE = (32768, 65535)
 
 
-def port_ranges():
+def port_ranges() -> List[Tuple[int, int]]:
     """
     Returns a list of ephemeral port ranges for current machine.
     """
@@ -30,22 +31,25 @@ def port_ranges():
     return [DEFAULT_EPHEMERAL_PORT_RANGE]
 
 
-def _linux_ranges():
+def _linux_ranges() -> List[Tuple[int, int]]:
     with open("/proc/sys/net/ipv4/ip_local_port_range") as f:
         # use readline() instead of read() for linux + musl
         low, high = f.readline().split()
         return [(int(low), int(high))]
 
 
-def _bsd_ranges():
+def _bsd_ranges() -> List[Tuple[int, int]]:
     pp = subprocess.Popen(
         ["sysctl", "net.inet.ip.portrange"], stdout=subprocess.PIPE
     )
     stdout, stderr = pp.communicate()
     lines = stdout.decode("ascii").split("\n")
-    out = dict(
+    out: Dict[str, str] = dict(
         [
-            [x.strip().rsplit(".")[-1] for x in line.split(":")]
+            [
+                x.strip().rsplit(".")[-1]  # type: ignore[misc]
+                for x in line.split(":")
+            ]
             for line in lines
             if line
         ]