-
Notifications
You must be signed in to change notification settings - Fork 3
/
docker-compose-check.py
executable file
·115 lines (90 loc) · 3.09 KB
/
docker-compose-check.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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
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
#!/usr/bin/env python3
# This code is available under the MIT license: https://opensource.org/licenses/MIT
from pathlib import Path
import subprocess
import json
from dataclasses import dataclass
from typing import List, Optional
container_hashes = subprocess.check_output(["docker", "ps", "-q"], text=True)
@dataclass
class Container:
container_hash: str
config_hash: str
project: str
config: str
containernum: int
service: str
work_dir: Path
@property
def name(self) -> str:
return f"{self.project}_{self.service}_{self.containernum}"
# start gist block
# https://gist.github.com/laundmo/2e2f7314570d2f86a4af8df4b1812b63
def text_width(text: str) -> int:
return len(max(text.split("\n"), key=len))
def max_text_width(texts: List[str]) -> int:
return len(max(texts, key=text_width))
def format_table(
table: List[List[str]],
headers: Optional[List[str]] = None,
justify=str.ljust,
) -> str:
if headers is None:
headers = [""] * len(table)
col_l = [max_text_width([headers[i]] + col) for i, col in enumerate(table)]
output = ""
if max(headers, key=len) != "":
output += (
" | ".join(justify(item, col_l[i]) for i, item in enumerate(headers)) + "\n"
)
output += " | ".join("-" * col_l[i] for i, _ in enumerate(table)) + "\n"
for row in zip(*table):
output += (
" | ".join(justify(item, col_l[i]) for i, item in enumerate(row)) + "\n"
)
return output
# end gist block
containers: List[Container] = []
for container_hash in container_hashes.strip().split("\n"):
labels_json = subprocess.check_output(
["docker", "inspect", container_hash, "-f", "'{{json .Config.Labels}}'"],
text=True,
)
labels_json = labels_json.strip().rstrip("'").lstrip("'")
labels = json.loads(labels_json)
c = Container(
container_hash,
labels["com.docker.compose.config-hash"],
labels["com.docker.compose.project"],
labels["com.docker.compose.project.config_files"],
int(labels["com.docker.compose.container-number"]),
labels["com.docker.compose.service"],
Path(labels["com.docker.compose.project.working_dir"]),
)
containers.append(c)
hashes = subprocess.check_output(
["docker-compose", "config", "--hash=*"], text=True
).strip()
confs = [srv_hash.split(" ") for srv_hash in hashes.split("\n")]
tbl: List[List[str]] = []
for service, conf_hash in confs:
workdir = Path(".")
service_conts = list(
filter(
lambda c: c.work_dir.resolve() == workdir.resolve()
and c.service == service,
containers,
)
)
col = [service]
if len(service_conts) == 0:
col += ["New container", ""]
else:
for result in service_conts:
if result.config_hash != conf_hash:
col += ["Restarting container", result.name]
else:
col += ["Keeping container", result.name]
tbl.append(col)
tbl = list(map(list, zip(*tbl)))
print(format_table(tbl, headers=["service", "todo", "container"]))