Add type hint to host folder

libp2p/host/basic_host.py

@@ -1,6 +1,33 @@
+from typing import (
+    Any,
+    Callable,
+    List,
+    Sequence,
+)
+
 import multiaddr
 
-from .host_interface import IHost
+from .host_interface import (
+    IHost,
+)
+from libp2p.network.swarm import (
+    Swarm
+)
+from libp2p.peer.id import (
+    ID,
+)
+from libp2p.peer.peerinfo import (
+    PeerInfo,
+)
+from libp2p.peer.peerstore import (
+    PeerStore,
+)
+from libp2p.network.stream.net_stream_interface import (
+    INetStream,
+)
+from libp2p.routing.kademlia.kademlia_peer_router import (
+    KadmeliaPeerRouter,
+)
 
 # Upon host creation, host takes in options,
 # including the list of addresses on which to listen.
@@ -10,48 +37,53 @@ from .host_interface import IHost
 
 class BasicHost(IHost):
 
+    _network: Swarm
+    router: KadmeliaPeerRouter
+    peerstore: PeerStore
+
     # default options constructor
-    def __init__(self, network, router=None):
+    def __init__(self, network: Swarm, router: KadmeliaPeerRouter=None) -> None:
         self._network = network
         self._router = router
         self.peerstore = self._network.peerstore
 
-    def get_id(self):
+    def get_id(self) -> ID:
         """
         :return: peer_id of host
         """
         return self._network.get_peer_id()
 
-    def get_network(self):
+    def get_network(self) -> Swarm:
         """
         :return: network instance of host
         """
         return self._network
 
-    def get_peerstore(self):
+    def get_peerstore(self) -> PeerStore:
         """
         :return: peerstore of the host (same one as in its network instance)
         """
         return self.peerstore
 
-    def get_mux(self):
+    # FIXME: Replace with correct return type
+    def get_mux(self) -> Any:
         """
         :return: mux instance of host
         """
 
-    def get_addrs(self):
+    def get_addrs(self) -> List[multiaddr.Multiaddr]:
         """
         :return: all the multiaddr addresses this host is listening too
         """
         p2p_part = multiaddr.Multiaddr('/p2p/{}'.format(self.get_id().pretty()))
 
-        addrs = []
+        addrs: List[multiaddr.Multiaddr] = []
         for transport in self._network.listeners.values():
             for addr in transport.get_addrs():
                 addrs.append(addr.encapsulate(p2p_part))
         return addrs
 
-    def set_stream_handler(self, protocol_id, stream_handler):
+    def set_stream_handler(self, protocol_id: str, stream_handler: Callable[[INetStream], None]) -> bool:
         """
         set stream handler for host
         :param protocol_id: protocol id used on stream
@@ -62,16 +94,15 @@ class BasicHost(IHost):
 
     # protocol_id can be a list of protocol_ids
     # stream will decide which protocol_id to run on
-    async def new_stream(self, peer_id, protocol_ids):
+    async def new_stream(self, peer_id: ID, protocol_ids: Sequence[str]) -> INetStream:
         """
         :param peer_id: peer_id that host is connecting
         :param protocol_id: protocol id that stream runs on
-        :return: true if successful
+        :return: stream: new stream created
         """
-        stream = await self._network.new_stream(peer_id, protocol_ids)
-        return stream
+        return await self._network.new_stream(peer_id, protocol_ids)
 
-    async def connect(self, peer_info):
+    async def connect(self, peer_info: PeerInfo) -> None:
         """
         connect ensures there is a connection between this host and the peer with
         given peer_info.peer_id. connect will absorb the addresses in peer_info into its internal

libp2p/host/host_interface.py

@@ -1,34 +1,63 @@
 from abc import ABC, abstractmethod
+from typing import (
+    Any,
+    Callable,
+    Coroutine,
+    List,
+    Sequence,
+)
+
+import multiaddr
+
+from libp2p.network.swarm import (
+    Swarm
+)
+from libp2p.peer.id import (
+    ID,
+)
+from libp2p.peer.peerinfo import (
+    PeerInfo,
+)
+from libp2p.peer.peerstore import (
+    PeerStore,
+)
+from libp2p.network.stream.net_stream_interface import (
+    INetStream,
+)
+from libp2p.routing.kademlia.kademlia_peer_router import (
+    KadmeliaPeerRouter,
+)
 
 
 class IHost(ABC):
 
     @abstractmethod
-    def get_id(self):
+    def get_id(self) -> ID:
         """
         :return: peer_id of host
         """
 
     @abstractmethod
-    def get_network(self):
+    def get_network(self) -> Swarm:
         """
         :return: network instance of host
         """
 
+    # FIXME: Replace with correct return type
     @abstractmethod
-    def get_mux(self):
+    def get_mux(self) -> Any:
         """
         :return: mux instance of host
         """
 
     @abstractmethod
-    def get_addrs(self):
+    def get_addrs(self) -> List[multiaddr.Multiaddr]:
         """
         :return: all the multiaddr addresses this host is listening too
         """
 
     @abstractmethod
-    def set_stream_handler(self, protocol_id, stream_handler):
+    def set_stream_handler(self, protocol_id: str, stream_handler: Callable[[INetStream], None]) -> bool:
         """
         set stream handler for host
         :param protocol_id: protocol id used on stream
@@ -39,15 +68,15 @@ class IHost(ABC):
     # protocol_id can be a list of protocol_ids
     # stream will decide which protocol_id to run on
     @abstractmethod
-    def new_stream(self, peer_id, protocol_ids):
+    def new_stream(self, peer_id: ID, protocol_ids: Sequence[str]) -> Coroutine[Any, Any, INetStream]:
         """
         :param peer_id: peer_id that host is connecting
         :param protocol_ids: protocol ids that stream can run on
-        :return: true if successful
+        :return: stream: new stream created
         """
 
     @abstractmethod
-    def connect(self, peer_info):
+    def connect(self, peer_info: PeerInfo) -> Coroutine[Any, Any, None]:
         """
         connect ensures there is a connection between this host and the peer with
         given peer_info.peer_id. connect will absorb the addresses in peer_info into its internal