daemon: added class variable type hinting to core.api.grpc

daemon/core/api/grpc/client.py

@@ -5,7 +5,7 @@ gRpc client for interfacing with CORE.
 import logging
 import threading
 from contextlib import contextmanager
-from typing import Any, Callable, Dict, Generator, Iterable, List
+from typing import Any, Callable, Dict, Generator, Iterable, List, Optional
 
 import grpc
 
@@ -108,7 +108,7 @@ class InterfaceHelper:
         :param ip6_prefix: ip6 prefix to use for generation
         :raises ValueError: when both ip4 and ip6 prefixes have not been provided
         """
-        self.prefixes = IpPrefixes(ip4_prefix, ip6_prefix)
+        self.prefixes: IpPrefixes = IpPrefixes(ip4_prefix, ip6_prefix)
 
     def create_interface(
         self, node_id: int, interface_id: int, name: str = None, mac: str = None
@@ -177,10 +177,10 @@ class CoreGrpcClient:
 
         :param address: grpc server address to connect to
         """
-        self.address = address
-        self.stub = None
-        self.channel = None
-        self.proxy = proxy
+        self.address: str = address
+        self.stub: Optional[core_pb2_grpc.CoreApiStub] = None
+        self.channel: Optional[grpc.Channel] = None
+        self.proxy: bool = proxy
 
     def start_session(
         self,

daemon/core/api/grpc/events.py

@@ -140,9 +140,9 @@ class EventStreamer:
         :param session: session to process events for
         :param event_types: types of events to process
         """
-        self.session = session
-        self.event_types = event_types
-        self.queue = Queue()
+        self.session: Session = session
+        self.event_types: Iterable[core_pb2.EventType] = event_types
+        self.queue: Queue = Queue()
         self.add_handlers()
 
     def add_handlers(self) -> None:

daemon/core/api/grpc/grpcutils.py

@@ -1,6 +1,6 @@
 import logging
 import time
-from typing import Any, Dict, List, Tuple, Type
+from typing import Any, Dict, List, Tuple, Type, Union
 
 import grpc
 import netaddr
@@ -190,7 +190,8 @@ def convert_value(value: Any) -> str:
 
 
 def get_config_options(
-    config: Dict[str, str], configurable_options: Type[ConfigurableOptions]
+    config: Dict[str, str],
+    configurable_options: Union[ConfigurableOptions, Type[ConfigurableOptions]],
 ) -> Dict[str, common_pb2.ConfigOption]:
     """
     Retrieve configuration options in a form that is used by the grpc server.
@@ -418,7 +419,7 @@ def service_configuration(session: Session, config: ServiceConfig) -> None:
         service.shutdown = tuple(config.shutdown)
 
 
-def get_service_configuration(service: Type[CoreService]) -> NodeServiceData:
+def get_service_configuration(service: CoreService) -> NodeServiceData:
     """
     Convenience for converting a service to service data proto.
 

daemon/core/api/grpc/server.py

@@ -6,7 +6,7 @@ import tempfile
 import threading
 import time
 from concurrent import futures
-from typing import Iterable, Type
+from typing import Iterable, Optional, Type
 
 import grpc
 from grpc import ServicerContext
@@ -131,9 +131,9 @@ class CoreGrpcServer(core_pb2_grpc.CoreApiServicer):
 
     def __init__(self, coreemu: CoreEmu) -> None:
         super().__init__()
-        self.coreemu = coreemu
-        self.running = True
-        self.server = None
+        self.coreemu: CoreEmu = coreemu
+        self.running: bool = True
+        self.server: Optional[grpc.Server] = None
         atexit.register(self._exit_handler)
 
     def _exit_handler(self) -> None: