Add type hints for better code clarity and linting

icmplib/models.py

@@ -26,6 +26,7 @@
     <https://www.gnu.org/licenses/>.
 '''
 
+from typing import List, Optional
 from .exceptions import *
 from .utils import random_byte_message
 
@@ -74,8 +75,16 @@ class ICMPRequest:
     __slots__ = '_destination', '_id', '_sequence', '_payload', \
                 '_payload_size', '_ttl', '_traffic_class', '_time'
 
-    def __init__(self, destination, id, sequence, payload=None,
-            payload_size=56, ttl=64, traffic_class=0):
+    def __init__(
+            self, 
+            destination: str, 
+            id: int, 
+            sequence: int, 
+            payload: Optional[bytes] = None,
+            payload_size: int = 56, 
+            ttl: int = 64, 
+            traffic_class: int = 0
+            ):
 
         if payload:
             payload_size = len(payload)
@@ -199,8 +208,17 @@ class ICMPReply:
     __slots__ = '_source', '_family', '_id', '_sequence', '_type', \
                 '_code', '_bytes_received', '_time'
 
-    def __init__(self, source, family, id, sequence, type, code,
-            bytes_received, time):
+    def __init__(
+            self, 
+            source: str, 
+            family: int, 
+            id: int, 
+            sequence: int, 
+            type: int, 
+            code: int,
+            bytes_received: int, 
+            time: float
+            ):
 
         self._source = source
         self._family = family
@@ -323,7 +341,12 @@ class Host:
     '''
     __slots__ = '_address', '_packets_sent', '_rtts'
 
-    def __init__(self, address, packets_sent, rtts):
+    def __init__(
+            self, 
+            address: str, 
+            packets_sent: int, 
+            rtts: List[float]
+            ):
         self._address = address
         self._packets_sent = packets_sent
         self._rtts = rtts
@@ -471,7 +494,13 @@ class Hop(Host):
     '''
     __slots__ = '_address', '_packets_sent', '_rtts', '_distance'
 
-    def __init__(self, address, packets_sent, rtts, distance):
+    def __init__(
+            self, 
+            address: str, 
+            packets_sent: int, 
+            rtts: List[float], 
+            distance: int
+            ):
         super().__init__(address, packets_sent, rtts)
         self._distance = distance
 

icmplib/multiping.py

@@ -27,13 +27,22 @@
 '''
 
 import asyncio
+from typing import List, Optional
 
 from .ping import async_ping
 
 
-async def async_multiping(addresses, count=2, interval=0.5, timeout=2,
-        concurrent_tasks=50, source=None, family=None, privileged=True,
-        **kwargs):
+async def async_multiping(
+        addresses: List[str], 
+        count: int = 2, 
+        interval: float = 0.5, 
+        timeout: float = 2,
+        concurrent_tasks: int = 50, 
+        source: Optional[str] = None, 
+        family: Optional[int] = None, 
+        privileged: Optional[bool] = True,
+        **kwargs
+        ):
     '''
     Send ICMP Echo Request packets to several network hosts.
 
@@ -163,9 +172,17 @@ async def async_multiping(addresses, count=2, interval=0.5, timeout=2,
     return [task.result() for task in tasks]
 
 
-def multiping(addresses, count=2, interval=0.5, timeout=2,
-        concurrent_tasks=50, source=None, family=None, privileged=True,
-        **kwargs):
+def multiping(
+        addresses: List[str], 
+        count: int = 2, 
+        interval: float = 0.5, 
+        timeout: float = 2,
+        concurrent_tasks: int = 50, 
+        source: Optional[str] = None, 
+        family: Optional[int] = None, 
+        privileged: Optional[bool] = True,
+        **kwargs
+        ):
     '''
     Send ICMP Echo Request packets to several network hosts.
 

icmplib/ping.py

@@ -28,15 +28,25 @@
 
 import asyncio
 from time import sleep
+from typing import Optional
 
 from .sockets import ICMPv4Socket, ICMPv6Socket, AsyncSocket
 from .models import ICMPRequest, Host
 from .exceptions import ICMPLibError
 from .utils import *
 
 
-def ping(address, count=4, interval=1, timeout=2, id=None, source=None,
-        family=None, privileged=True, **kwargs):
+def ping(
+        address: str, 
+        count: int = 4, 
+        interval: float = 1, 
+        timeout: float = 2, 
+        id: Optional[int] = None, 
+        source: Optional[str] = None,
+        family: Optional[int] = None, 
+        privileged: Optional[bool] = True, 
+        **kwargs
+        ):
     '''
     Send ICMP Echo Request packets to a network host.
 
@@ -165,8 +175,10 @@ def ping(address, count=4, interval=1, timeout=2, id=None, source=None,
     return Host(address, packets_sent, rtts)
 
 
-async def async_ping(address, count=4, interval=1, timeout=2, id=None,
-        source=None, family=None, privileged=True, **kwargs):
+async def async_ping(address: str, count: int = 4, interval: float = 1, 
+                     timeout: float = 2, id: Optional[int] = None, 
+                     source: Optional[str] = None, family: Optional[int] = None, 
+                     privileged: Optional[bool] = True, **kwargs):
     '''
     Send ICMP Echo Request packets to a network host.
 

icmplib/sockets.py

@@ -29,8 +29,9 @@
 import socket, asyncio
 from struct import pack, unpack
 from time import time
+from typing import Optional
 
-from .models import ICMPReply
+from .models import ICMPReply, ICMPRequest
 from .exceptions import *
 from .utils import PLATFORM_LINUX, PLATFORM_MACOS, PLATFORM_WINDOWS
 
@@ -74,7 +75,7 @@ class ICMPSocket:
     _ICMP_ECHO_REQUEST       = -1
     _ICMP_ECHO_REPLY         = -1
 
-    def __init__(self, address=None, privileged=True):
+    def __init__(self, address: Optional[str] = None, privileged: Optional[bool] = True):
         self._sock = None
         self._address = address
 
@@ -233,7 +234,7 @@ def _parse_reply(self, packet, source, current_time):
             bytes_received=bytes_received,
             time=current_time)
 
-    def send(self, request):
+    def send(self, request: ICMPRequest):
         '''
         Send an ICMP request message over the network to a remote host.
 
@@ -287,7 +288,7 @@ def send(self, request):
         except OSError as err:
             raise ICMPSocketError(str(err))
 
-    def receive(self, request=None, timeout=2):
+    def receive(self, request: Optional[ICMPRequest] = None, timeout: float = 2):
         '''
         Receive an ICMP reply message from the socket.
 
@@ -532,7 +533,7 @@ def broadcast(self):
             socket.SO_BROADCAST) > 0
 
     @broadcast.setter
-    def broadcast(self, enable):
+    def broadcast(self, enable: bool):
         self._sock.setsockopt(
             socket.SOL_SOCKET,
             socket.SO_BROADCAST,
@@ -666,7 +667,7 @@ class AsyncSocket:
     '''
     __slots__ = '_icmp_sock'
 
-    def __init__(self, icmp_sock):
+    def __init__(self, icmp_sock: ICMPSocket):
         self._icmp_sock = icmp_sock
         self._icmp_sock.blocking = False
 
@@ -701,7 +702,7 @@ def __del__(self):
         '''
         self.close()
 
-    async def receive(self, request=None, timeout=2):
+    async def receive(self, request: Optional[ICMPRequest] = None, timeout: float = 2):
         '''
         Receive an ICMP reply message from the socket.
 

icmplib/traceroute.py

@@ -27,16 +27,27 @@
 '''
 
 from time import sleep
+from typing import Optional
 
 from .sockets import ICMPv4Socket, ICMPv6Socket
 from .models import ICMPRequest, Hop
 from .exceptions import TimeExceeded, ICMPLibError
 from .utils import *
 
 
-def traceroute(address, count=2, interval=0.05, timeout=2, first_hop=1,
-        max_hops=30, fast=False, id=None, source=None, family=None,
-        **kwargs):
+def traceroute(
+        address: str, 
+        count: int = 2, 
+        interval: float = 0.05, 
+        timeout: float = 2, 
+        first_hop: int = 1,
+        max_hops: int = 30, 
+        fast: Optional[bool] = False, 
+        id: Optional[int] = None,
+        source: Optional[str] = None, 
+        family: Optional[int] = None,
+        **kwargs
+        ):
     '''
     Determine the route to a destination host.
 

icmplib/utils.py

@@ -33,6 +33,7 @@
 from os import getpid
 from re import match
 from random import choices
+from typing import Optional
 
 from .exceptions import NameLookupError
 
@@ -46,7 +47,7 @@
 _current_id = PID
 
 
-def random_byte_message(size):
+def random_byte_message(size: int):
     '''
     Generate a random byte sequence of the specified size.
 
@@ -74,7 +75,7 @@ def unique_identifier():
         return _current_id
 
 
-def resolve(name, family=None):
+def resolve(name, family: Optional[int] = None):
     '''
     Resolve a hostname or FQDN to an IP address. Depending on the name
     specified in parameters, several IP addresses may be returned.
@@ -120,7 +121,7 @@ def resolve(name, family=None):
     raise NameLookupError(name)
 
 
-async def async_resolve(name, family=None):
+async def async_resolve(name, family: Optional[int] = None):
     '''
     Resolve a hostname or FQDN to an IP address. Depending on the name
     specified in parameters, several IP addresses may be returned.
@@ -170,7 +171,7 @@ async def async_resolve(name, family=None):
     raise NameLookupError(name)
 
 
-def is_hostname(name):
+def is_hostname(name: str):
     '''
     Indicate whether the specified name is a hostname or an FQDN.
     Return a `boolean`.
@@ -180,7 +181,7 @@ def is_hostname(name):
     return match(pattern, name) is not None
 
 
-def is_ipv4_address(address):
+def is_ipv4_address(address: str):
     '''
     Indicate whether the specified address is an IPv4 address.
     Return a `boolean`.
@@ -190,7 +191,7 @@ def is_ipv4_address(address):
     return match(pattern, address) is not None
 
 
-def is_ipv6_address(address):
+def is_ipv6_address(address: str):
     '''
     Indicate whether the specified address is an IPv6 address.
     Return a `boolean`.