Skip to content

Module: docker

destroy_services(dir)

Stop and remove all service containers found in the config file.

Source code in src/infernet_cli/cli/docker.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def destroy_services(dir: str) -> None:
    """Stop and remove all service containers found in the config file."""

    if not os.path.exists(f"{dir}/config.json"):
        raise click.ClickException(f"{dir}/config.json not found.")

    # Read container IDs from config.json
    config = json.load(open(f"{dir}/config.json"))
    container_ids = [container["id"] for container in config["containers"]]

    # Force stop and remove containers
    for container_id in container_ids:
        # Stop and remove the container
        run_command(["docker", "stop", container_id])
        run_command(["docker", "rm", container_id])

docker_destroy(dir)

Destroy the containers.

Source code in src/infernet_cli/cli/docker.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def docker_destroy(dir: str) -> None:
    """Destroy the containers."""
    try:
        # Run the docker compose command without showing the output
        result = run_command(
            ["docker", "compose", "-f", f"{dir}/docker-compose.yaml", "rm", "-f"],
        )

        # Check if the command was successful
        if result.returncode == 0:
            click.echo("Containers destroyed successfully.")
        else:
            click.echo(f"Failed to destroy containers. Error: {result.stderr}")

    except Exception as e:
        click.echo(f"An error occurred while destroying containers: {e}")

docker_start(dir)

Start the containers.

Source code in src/infernet_cli/cli/docker.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def docker_start(dir: str) -> None:
    """Start the containers."""
    try:
        # Run the docker compose command without showing the output
        result = run_command(
            ["docker", "compose", "-f", f"{dir}/docker-compose.yaml", "up", "-d"]
        )

        # Check if the command was successful
        if result.returncode == 0:
            click.echo("Containers started successfully.")
        else:
            click.echo(f"Failed to start containers. Error: {result.stderr}")

    except Exception as e:
        click.echo(f"An error occurred: {e}")

docker_stop(dir)

Stop the containers.

Source code in src/infernet_cli/cli/docker.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def docker_stop(dir: str) -> None:
    """Stop the containers."""
    try:
        # Run the docker compose command without showing the output
        result = run_command(
            ["docker", "compose", "-f", f"{dir}/docker-compose.yaml", "stop"],
        )

        # Check if the command was successful
        if result.returncode == 0:
            click.echo("Containers stopped successfully.")
        else:
            click.echo(f"Failed to stop containers. Error: {result.stderr}")

    except Exception as e:
        click.echo(f"An error occurred while stopping containers: {e}")

health_check(dir)

Check if all containers are up and healthy.

Source code in src/infernet_cli/cli/docker.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def health_check(dir: str) -> None:
    """Check if all containers are up and healthy."""
    try:
        # Run `docker compose ps` to get the status of the containers
        result = run_command(
            [
                "docker",
                "compose",
                "-f",
                f"{dir}/docker-compose.yaml",
                "ps",
                "-a",
                "--format",
                "json",
            ],
        )
        # Parse the result as JSON to inspect each container
        containers = ",".join(result.stdout.split("\n")[:-1])
        containers_obj = json.loads(f"[{containers}]")

        if len(containers_obj) == 0:
            click.echo("No containers found.")
            return

        # Check if the command was successful
        if result.returncode != 0:
            click.echo(f"Failed to get the container status. Error: {result.stderr}")
            return

        # Check if all containers are up and healthy
        all_healthy = True
        for container in containers_obj:
            name = container["Service"]
            status = container["State"]  # State includes "running", "exited", etc.

            # Simply check if the container is running
            if status != "running":
                click.echo(f"Container {name} is not running. Status: {status}.")
                all_healthy = False

        if all_healthy:
            click.echo("\033[92mAll containers are up and running.\033[0m")
        else:
            click.echo("\033[91mSome containers are not running.\033[0m")

    except Exception as e:
        click.echo(f"An error occurred while checking container health: {e}")

run_command(command)

Helper function to run shell commands.

Source code in src/infernet_cli/cli/docker.py 
 8
 9
10
11
12
13
14
15
def run_command(command: list[str]) -> subprocess.CompletedProcess[str]:
    """Helper function to run shell commands."""
    return subprocess.run(
        command,
        stdout=subprocess.PIPE,  # Suppress standard output
        stderr=subprocess.PIPE,  # Suppress standard error
        text=True,  # Get the output as text
    )