add more type hints to docker_util.py

simulation-system/libs/csle-common/src/csle_common/constants/constants.py

@@ -1430,6 +1430,7 @@ class DOCKER:
     IMAGE_CREATED = "Created"
     IMAGE_OS = "Os"
     IMAGE_SIZE = "Size"
+    IMAGE = "Image"
     IMAGE_ARCHITECTURE = "Architecture"
     REPO_TAGS = "RepoTags"
     ENTRYPOINT = "Entrypoint"

simulation-system/libs/csle-common/src/csle_common/dao/docker/docker_container_metadata.py

@@ -1,4 +1,5 @@
 from typing import Dict, Any
+from docker.models.containers import Container
 from csle_base.json_serializable import JSONSerializable
 
 
@@ -10,9 +11,9 @@ class DockerContainerMetadata(JSONSerializable):
     def __init__(self, name: str, status: str, short_id: str, image_short_id: str, image_tags: list, id: str,
                  created: str, ip: str, network_id: int, gateway: str, mac: str, ip_prefix_len: int,
                  name2: str, level: str, hostname: str, image_name: str, net: str,
-                 dir: str, config_path: str, container_handle: str, emulation: str, kafka_container: str):
+                 dir: str, config_path: str, container_handle: Container, emulation: str, kafka_container: str) -> None:
         """
-        Intializes the DTO
+        Initializes the DTO
 
         :param name: the name of the container
         :param status: the status of the container

simulation-system/libs/csle-common/src/csle_common/util/docker_util.py

@@ -1,5 +1,6 @@
 from typing import List, Union
 import docker
+from docker.models.containers import Container
 import os
 import json
 from csle_common.dao.docker.docker_container_metadata import DockerContainerMetadata
@@ -20,13 +21,11 @@ def parse_runnning_emulation_infos() -> List[DockerEnvMetadata]:
 
         :return: a list of environment DTOs
         """
-        client_1 = docker.from_env()
-        client_2 = docker.APIClient(base_url=constants.DOCKER.UNIX_DOCKER_SOCK_URL)
+        client_1: docker.DockerClient = docker.from_env()
+        client_2: docker.APIClient = docker.APIClient(base_url=constants.DOCKER.UNIX_DOCKER_SOCK_URL)
         parsed_containers = DockerUtil.parse_running_containers(client_1=client_1, client_2=client_2)
-        for container in parsed_containers:
-            if container is None:
-                raise ValueError("contianer is None")
-        emulations: List[str] = list(set(list(map(lambda x: x.emulation, filter(lambda x: x is not None, parsed_containers)))))
+        emulations: List[str] = list(set(list(map(lambda x: x.emulation, filter(lambda x: x is not None,
+                                                                                parsed_containers)))))
         parsed_envs = DockerUtil.parse_running_emulation_envs(emulations=emulations, containers=parsed_containers)
         return parsed_envs
 
@@ -37,38 +36,41 @@ def get_container_hex_id(name: str) -> Union[str, None]:
 
         :return: the id
         """
-        client_1 = docker.from_env()
-        client_2 = docker.APIClient(base_url=constants.DOCKER.UNIX_DOCKER_SOCK_URL)
-        containers = DockerUtil.parse_running_containers(client_1=client_1, client_2=client_2)
+        client_1: docker.DockerClient = docker.from_env()
+        client_2: docker.APIClient = docker.APIClient(base_url=constants.DOCKER.UNIX_DOCKER_SOCK_URL)
+        containers: List[DockerContainerMetadata] = DockerUtil.parse_running_containers(client_1=client_1,
+                                                                                        client_2=client_2)
         for container in containers:
             if container.name == name:
                 return container.id
         return None
 
     @staticmethod
-    def parse_running_containers(client_1, client_2) -> List[DockerContainerMetadata]:
+    def parse_running_containers(client_1: docker.DockerClient, client_2: docker.APIClient) \
+            -> List[DockerContainerMetadata]:
         """
         Queries docker to get a list of all running containers
 
         :param client_1: docker client 1
         :param client_2:  docker client 2
         :return: list of parsed running containers
         """
-        containers = client_1.containers.list()
+        containers: List[Container] = client_1.containers.list()
         parsed_containers = DockerUtil.parse_containers(containers=containers, client2=client_2)
         return parsed_containers
 
     @staticmethod
-    def parse_stopped_containers(client_1, client2) -> List[DockerContainerMetadata]:
+    def parse_stopped_containers(client_1: docker.DockerClient, client2: docker.APIClient) \
+            -> List[DockerContainerMetadata]:
         """
         Queries docker to get a list of all stopped csle containers
 
         :param client_1: docker client 1
         :param client2: docker client 2
         :return: list of parsed containers
         """
-        containers = client_1.containers.list(all=True)
-        stopped_containers = list(filter(
+        containers: List[docker.models.containers.Container] = client_1.containers.list(all=True)
+        stopped_containers: List[docker.models.containers.Container] = list(filter(
             lambda x: (x.status == constants.DOCKER.CONTAINER_EXIT_STATUS
                        or x.status == constants.DOCKER.CONTAINER_CREATED_STATUS), containers))
         parsed_containers = DockerUtil.parse_containers(containers=stopped_containers, client2=client2)
@@ -103,7 +105,8 @@ def parse_running_emulation_envs(emulations: List[str], containers: List[DockerC
         return parsed_envs
 
     @staticmethod
-    def parse_containers(containers, client2) -> List[DockerContainerMetadata]:
+    def parse_containers(containers: List[docker.models.containers.Container], client2: docker.APIClient) \
+            -> List[DockerContainerMetadata]:
         """
         Queries docker to get a list of running or stopped csle containers
 
@@ -122,10 +125,10 @@ def parse_containers(containers, client2) -> List[DockerContainerMetadata]:
                 if len(list(inspect_info[constants.DOCKER.NETWORK_SETTINGS][constants.DOCKER.NETWORKS].keys())) > 0:
                     net = list(inspect_info[constants.DOCKER.NETWORK_SETTINGS][constants.DOCKER.NETWORKS].keys())[0]
                 labels = c.labels
-                config_path = None
-                dir_path = None
-                emulation = None
-                kafka_config = None
+                config_path = ""
+                dir_path = ""
+                emulation = ""
+                kafka_config = ""
                 if constants.DOCKER.CFG in labels:
                     config_path = labels[constants.DOCKER.CFG]
                 if constants.DOCKER.CONTAINER_CONFIG_DIR in labels:
@@ -134,7 +137,6 @@ def parse_containers(containers, client2) -> List[DockerContainerMetadata]:
                     emulation = labels[constants.DOCKER.EMULATION]
                 if constants.DOCKER.KAFKA_CONFIG in labels:
                     kafka_config = labels[constants.DOCKER.KAFKA_CONFIG]
-
                 ip = ""
                 network_id = -1
                 gateway = ""
@@ -151,20 +153,14 @@ def parse_containers(containers, client2) -> List[DockerContainerMetadata]:
                         constants.DOCKER.MAC_ADDRESS_INFO]
                     ip_prefix_len = inspect_info[constants.DOCKER.NETWORK_SETTINGS][constants.DOCKER.NETWORKS][net][
                         constants.DOCKER.IP_PREFIX_LEN_INFO]
-                if emulation is None:
-                    raise ValueError("emulation is None, cannot be None")
                 parsed_c = DockerContainerMetadata(
                     name=c.name, status=c.status, short_id=c.short_id, image_short_id=c.image.short_id,
                     image_tags=c.image.tags, id=c.id,
                     created=inspect_info[constants.DOCKER.CREATED_INFO],
-                    ip=ip,
-                    network_id=network_id,
-                    gateway=gateway,
-                    mac=mac,
-                    ip_prefix_len=ip_prefix_len,
+                    ip=ip, network_id=network_id, gateway=gateway, mac=mac, ip_prefix_len=ip_prefix_len,
                     name2=container_name_2, level=level,
                     hostname=inspect_info[constants.DOCKER.CONFIG][constants.DOCKER.HOSTNAME_INFO],
-                    image_name=inspect_info[constants.DOCKER.CONFIG]["Image"],
+                    image_name=inspect_info[constants.DOCKER.CONFIG][constants.DOCKER.IMAGE],
                     net=net, dir=dir_path, config_path=config_path,
                     container_handle=c, emulation=emulation, kafka_container=kafka_config)
                 parsed_containers.append(parsed_c)