Skip to content

node_deployer.utils

Singleton

Bases: type

A singleton metaclass

Source code in src/node_deployer/utils.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class Singleton(type):
    """A singleton metaclass"""

    _instance = None

    def __call__(cls, *args, **kwargs):
        """Creates a new instance of the class if one does not already exist

        Returns:
            cls._instance: The instance of the class
        """
        if cls._instance is None:
            cls._instance = super().__call__(*args, **kwargs)
        return cls._instance

__call__(*args, **kwargs)

Creates a new instance of the class if one does not already exist

Returns:

Type Description

cls._instance: The instance of the class
Source code in src/node_deployer/utils.py 
32
33
34
35
36
37
38
39
40
def __call__(cls, *args, **kwargs):
    """Creates a new instance of the class if one does not already exist

    Returns:
        cls._instance: The instance of the class
    """
    if cls._instance is None:
        cls._instance = super().__call__(*args, **kwargs)
    return cls._instance

ensure_build_dir(f)

Ensures that the build directory exists before running the decorated function

Parameters:

Name Type Description Default
f Callable

The function to decorate

 required

Returns:

Name Type Description
Callable Callable

The decorated function
Source code in src/node_deployer/utils.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def ensure_build_dir(f: Callable) -> Callable:
    """Ensures that the build directory exists before running the decorated function

    Args:
        f (Callable): The function to decorate

    Returns:
        Callable: The decorated function
    """

    @wraps(f)
    def wrapper(*args, **kwargs):
        Path(config.BUILD_DIR).mkdir(exist_ok=True, parents=True)
        return f(*args, **kwargs)

    return wrapper

next_free_tcp_port(port)

Finds the next free port after the specified port

Parameters:

Name Type Description Default
port int

The port to start searching from

 required

Raises:

Type Description
ValueError

If no free ports are found

Returns:

Name Type Description
int int

The next free port
Source code in src/node_deployer/utils.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def next_free_tcp_port(port: int) -> int:
    """Finds the next free port after the specified port

    Args:
        port (int): The port to start searching from

    Raises:
        ValueError: If no free ports are found

    Returns:
        int: The next free port
    """
    ports: List[int] = []
    try:
        containers = config.CLIENT.containers.list(all=True)
        ports = []
        for container in containers:
            port_values = container.ports.values()
            if not port_values:
                continue
            for x in list(container.ports.values())[0]:
                ports.append(int(x["HostPort"]))
    except docker.errors.NotFound:  # type: ignore
        # * This error is raised if container list changes between getting the list and
        # * getting the ports. If this happens, just try again
        return next_free_tcp_port(port)
    if not ports:
        return port
    unique_ports = set(ports)
    while port in unique_ports:
        port += 1
        if port > 65535:
            raise ValueError("No free ports")
    return port
Last update: November 7, 2023
Created: November 7, 2023