264 lines
8.1 KiB
Python
264 lines
8.1 KiB
Python
"""Docker CLI wrapper — no SDK needed."""
|
|
import json, os, re, subprocess, shutil, threading
|
|
from pathlib import Path
|
|
|
|
COMPOSE_NAMES = ("compose.yml", "compose.yaml", "docker-compose.yml", "docker-compose.yaml")
|
|
|
|
|
|
def _bin() -> str:
|
|
for p in ("/usr/local/bin/docker", "/usr/bin/docker"):
|
|
if os.path.isfile(p):
|
|
return p
|
|
return shutil.which("docker") or "docker"
|
|
|
|
|
|
def _run(cmd, **kw) -> subprocess.CompletedProcess:
|
|
kw.setdefault("capture_output", True)
|
|
kw.setdefault("text", True)
|
|
kw.setdefault("timeout", 60)
|
|
try:
|
|
return subprocess.run(cmd, **kw)
|
|
except FileNotFoundError:
|
|
return subprocess.CompletedProcess(cmd, 127, "", "docker niet gevonden")
|
|
except subprocess.TimeoutExpired:
|
|
return subprocess.CompletedProcess(cmd, 1, "", "timeout")
|
|
|
|
|
|
def _compose_base() -> list[str]:
|
|
# Probeer docker compose (plugin)
|
|
r = _run([_bin(), "compose", "version"], timeout=5)
|
|
if r.returncode == 0:
|
|
return [_bin(), "compose"]
|
|
# Probeer standalone docker-compose
|
|
dc = shutil.which("docker-compose")
|
|
if dc:
|
|
return [dc]
|
|
# Probeer compose plugin op alternatieve locaties
|
|
for plugin_dir in ("/usr/lib/docker/cli-plugins",
|
|
"/usr/libexec/docker/cli-plugins",
|
|
"/usr/local/lib/docker/cli-plugins",
|
|
os.path.expanduser("~/.docker/cli-plugins")):
|
|
compose_bin = os.path.join(plugin_dir, "docker-compose")
|
|
if os.path.isfile(compose_bin):
|
|
return [compose_bin]
|
|
# Laatste poging: docker-compose uit PATH
|
|
return [_bin(), "compose"]
|
|
|
|
|
|
# Cache compose base command (thread-safe)
|
|
_compose_cmd = None
|
|
_compose_lock = threading.Lock()
|
|
|
|
|
|
def _get_compose() -> list[str]:
|
|
global _compose_cmd
|
|
if _compose_cmd is None:
|
|
with _compose_lock:
|
|
if _compose_cmd is None:
|
|
_compose_cmd = _compose_base()
|
|
return _compose_cmd
|
|
|
|
|
|
def _stream(cmd, cwd=None, log_fn=None) -> int:
|
|
try:
|
|
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
text=True, cwd=cwd)
|
|
for line in p.stdout:
|
|
s = line.rstrip()
|
|
if s and log_fn:
|
|
log_fn(s)
|
|
return p.wait()
|
|
except Exception as e:
|
|
if log_fn:
|
|
log_fn(f"Fout: {e}")
|
|
return 1
|
|
|
|
|
|
# ── Info ──────────────────────────────────────────────────────────────────────
|
|
|
|
def info() -> dict:
|
|
r = _run([_bin(), "info", "--format", "json"], timeout=10)
|
|
if r.returncode == 0:
|
|
try:
|
|
d = json.loads(r.stdout)
|
|
return {"ok": True, "version": d.get("ServerVersion", "?"),
|
|
"containers": d.get("Containers", 0),
|
|
"running": d.get("ContainersRunning", 0),
|
|
"images": d.get("Images", 0)}
|
|
except Exception:
|
|
pass
|
|
# Fallback
|
|
ver = _run([_bin(), "version", "--format", "{{.Server.Version}}"], timeout=5)
|
|
ps = _run([_bin(), "ps", "-q"], timeout=5)
|
|
imgs = _run([_bin(), "images", "-q"], timeout=5)
|
|
return {
|
|
"ok": ver.returncode == 0,
|
|
"version": ver.stdout.strip() if ver.returncode == 0 else "?",
|
|
"running": len(ps.stdout.strip().splitlines()) if ps.returncode == 0 else 0,
|
|
"containers": 0,
|
|
"images": len(imgs.stdout.strip().splitlines()) if imgs.returncode == 0 else 0,
|
|
}
|
|
|
|
|
|
# ── Images ────────────────────────────────────────────────────────────────────
|
|
|
|
def images() -> list[dict]:
|
|
fmt = "{{.ID}}\t{{.Repository}}\t{{.Tag}}\t{{.Size}}\t{{.CreatedSince}}"
|
|
r = _run([_bin(), "images", "--format", fmt], timeout=15)
|
|
if r.returncode != 0:
|
|
return []
|
|
out = []
|
|
for line in r.stdout.strip().splitlines():
|
|
p = line.split("\t")
|
|
if len(p) >= 4:
|
|
out.append({"id": p[0], "repo": p[1], "tag": p[2],
|
|
"size": p[3], "age": p[4] if len(p) > 4 else ""})
|
|
return out
|
|
|
|
|
|
def rmi(image_id: str, force=False) -> tuple[bool, str]:
|
|
cmd = [_bin(), "rmi"]
|
|
if force:
|
|
cmd.append("-f")
|
|
cmd.append(image_id)
|
|
r = _run(cmd, timeout=30)
|
|
return r.returncode == 0, (r.stderr or r.stdout).strip()
|
|
|
|
|
|
def prune_images() -> tuple[bool, str]:
|
|
r = _run([_bin(), "image", "prune", "-af"], timeout=120)
|
|
return r.returncode == 0, (r.stdout or r.stderr).strip()
|
|
|
|
|
|
# ── Container management ─────────────────────────────────────────────────────
|
|
|
|
def restart_container(name: str) -> tuple[bool, str]:
|
|
r = _run([_bin(), "restart", name], timeout=30)
|
|
return r.returncode == 0, (r.stderr or r.stdout).strip()
|
|
|
|
|
|
# ── Compose helpers ───────────────────────────────────────────────────────────
|
|
|
|
def find_compose(d: Path) -> Path | None:
|
|
for n in COMPOSE_NAMES:
|
|
f = d / n
|
|
if f.exists():
|
|
return f
|
|
return None
|
|
|
|
|
|
def has_compose(d: Path) -> bool:
|
|
return find_compose(d) is not None
|
|
|
|
|
|
def compose_ps(d: Path, name=None) -> list[dict]:
|
|
f = find_compose(d)
|
|
if not f:
|
|
return []
|
|
cmd = _get_compose() + ["-f", str(f)]
|
|
if name:
|
|
cmd += ["-p", name]
|
|
cmd += ["ps", "--format", "json"]
|
|
r = _run(cmd, cwd=str(d), timeout=10)
|
|
if r.returncode != 0:
|
|
return []
|
|
out = []
|
|
for line in r.stdout.strip().splitlines():
|
|
try:
|
|
c = json.loads(line)
|
|
out.append({
|
|
"name": c.get("Name", ""),
|
|
"service": c.get("Service", ""),
|
|
"state": c.get("State", ""),
|
|
"status": c.get("Status", ""),
|
|
"ports": c.get("Ports", ""),
|
|
"running": c.get("State", "").lower() in ("running", "up"),
|
|
})
|
|
except Exception:
|
|
pass
|
|
return out
|
|
|
|
|
|
def compose_up(d: Path, log_fn=None, name=None) -> int:
|
|
f = find_compose(d)
|
|
if not f:
|
|
return 1
|
|
cmd = _get_compose() + ["-f", str(f)]
|
|
if name:
|
|
cmd += ["-p", name]
|
|
cmd += ["up", "-d", "--remove-orphans"]
|
|
return _stream(cmd, cwd=str(d), log_fn=log_fn)
|
|
|
|
|
|
def compose_down(d: Path, log_fn=None, volumes=False, name=None) -> int:
|
|
f = find_compose(d)
|
|
if not f:
|
|
return 1
|
|
cmd = _get_compose() + ["-f", str(f)]
|
|
if name:
|
|
cmd += ["-p", name]
|
|
cmd += ["down"]
|
|
if volumes:
|
|
cmd.append("-v")
|
|
return _stream(cmd, cwd=str(d), log_fn=log_fn)
|
|
|
|
|
|
def compose_pull(d: Path, log_fn=None) -> int:
|
|
f = find_compose(d)
|
|
if not f:
|
|
return 1
|
|
cmd = _get_compose() + ["-f", str(f), "pull"]
|
|
return _stream(cmd, cwd=str(d), log_fn=log_fn)
|
|
|
|
|
|
def compose_logs(d: Path, tail=80, name=None) -> str:
|
|
f = find_compose(d)
|
|
if not f:
|
|
return ""
|
|
cmd = _get_compose() + ["-f", str(f)]
|
|
if name:
|
|
cmd += ["-p", name]
|
|
cmd += ["logs", "--tail", str(tail), "--no-color"]
|
|
r = _run(cmd, cwd=str(d), timeout=15)
|
|
return r.stdout if r.returncode == 0 else r.stderr
|
|
|
|
|
|
def read_env(d: Path) -> str:
|
|
f = d / ".env"
|
|
return f.read_text() if f.exists() else ""
|
|
|
|
|
|
def write_env(d: Path, content: str):
|
|
f = d / ".env"
|
|
f.write_text(content)
|
|
|
|
|
|
def read_compose(d: Path) -> str:
|
|
f = find_compose(d)
|
|
return f.read_text() if f else ""
|
|
|
|
|
|
def write_compose(d: Path, content: str):
|
|
f = find_compose(d)
|
|
if f:
|
|
f.write_text(content)
|
|
|
|
|
|
def used_ports() -> set[int]:
|
|
r = _run([_bin(), "ps", "-a", "--format", "{{.Ports}}"], timeout=10)
|
|
ports = set()
|
|
if r.returncode == 0:
|
|
for m in re.findall(r"(?:0\.0\.0\.0|::):(\d+)->", r.stdout):
|
|
try:
|
|
ports.add(int(m))
|
|
except ValueError:
|
|
pass
|
|
return ports
|
|
|
|
|
|
def next_free_port(start=8100) -> int:
|
|
used = used_ports()
|
|
p = start
|
|
while p in used:
|
|
p += 1
|
|
return p
|