phi/docker/resource/network.py

from typing import Optional, Any, List, Dict

from phi.docker.api_client import DockerApiClient
from phi.docker.resource.base import DockerResource
from phi.utils.log import logger


class DockerNetwork(DockerResource):
    resource_type: str = "Network"

    # driver (str) – Name of the driver used to create the network
    driver: Optional[str] = None
    # options (dict) – Driver options as a key-value dictionary
    options: Optional[Dict[str, Any]] = None
    # check_duplicate (bool) – Request daemon to check for networks with same name. Default: None.
    auto_remove: Optional[bool] = None
    # internal (bool) – Restrict external access to the network. Default False.
    internal: Optional[bool] = None
    # labels (dict) – Map of labels to set on the network. Default None.
    labels: Optional[Dict[str, Any]] = None
    # enable_ipv6 (bool) – Enable IPv6 on the network. Default False.
    enable_ipv6: Optional[bool] = None
    # attachable (bool) – If enabled, and the network is in the global scope
    # non-service containers on worker nodes will be able to connect to the network.
    attachable: Optional[bool] = None
    # scope (str) – Specify the network’s scope (local, global or swarm)
    scope: Optional[str] = None
    # ingress (bool) – If set, create an ingress network which provides the routing-mesh in swarm mode.
    ingress: Optional[bool] = None

    # Set skip_delete=True so that the network is not deleted when the `phi ws down` command is run
    skip_delete: bool = True
    skip_update: bool = True

    def _create(self, docker_client: DockerApiClient) -> bool:
        """Creates the Network on docker

        Args:
            docker_client: The DockerApiClient for the current cluster
        """
        from docker import DockerClient
        from docker.models.networks import Network

        logger.debug("Creating: {}".format(self.get_resource_name()))
        network_name: Optional[str] = self.name
        network_object: Optional[Network] = None

        try:
            _api_client: DockerClient = docker_client.api_client
            network_object = _api_client.networks.create(network_name)
            if network_object is not None:
                logger.debug("Network Created: {}".format(network_object.name))
            else:
                logger.debug("Network could not be created")
            # logger.debug("Network {}".format(network_object.attrs))
        except Exception:
            raise

        # By this step the network should be created
        # Validate that the network is created
        logger.debug("Validating network is created")
        if network_object is not None:
            # TODO: validate that the network was actually created
            self.active_resource = network_object
            return True

        logger.debug("Network not found")
        return False

    def _read(self, docker_client: DockerApiClient) -> Any:
        """Returns a Network object if the network is active

        Args:
            docker_client: The DockerApiClient for the current cluster"""
        from docker import DockerClient
        from docker.models.networks import Network

        logger.debug("Reading: {}".format(self.get_resource_name()))
        # Get active networks from the docker_client
        network_name: Optional[str] = self.name
        try:
            _api_client: DockerClient = docker_client.api_client
            network_list: Optional[List[Network]] = _api_client.networks.list()
            # logger.debug("network_list: {}".format(network_list))
            if network_list is not None:
                for network in network_list:
                    if network.name == network_name:
                        logger.debug(f"Network {network_name} exists")
                        self.active_resource = network
                        return network
        except Exception:
            logger.debug(f"Network {network_name} not found")

        return None

    def _delete(self, docker_client: DockerApiClient) -> bool:
        """Deletes the Network from docker

        Args:
            docker_client: The DockerApiClient for the current cluster
        """
        from docker.models.networks import Network
        from docker.errors import NotFound

        logger.debug("Deleting: {}".format(self.get_resource_name()))
        network_object: Optional[Network] = self._read(docker_client)
        # Return True if there is no Network to delete
        if network_object is None:
            return True

        # Delete Network
        try:
            self.active_resource = None
            network_object.remove()
        except Exception as e:
            logger.exception("Error while deleting network: {}".format(e))

        # Validate that the network is deleted
        logger.debug("Validating network is deleted")
        try:
            logger.debug("Reloading network_object: {}".format(network_object))
            network_object.reload()
        except NotFound:
            logger.debug("Got NotFound Exception, Network is deleted")
            return True

        return False