72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
"""Portainer API client."""
|
|
|
|
import json
|
|
import logging
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
PORTAINER_URL = "http://100.83.230.112:10000"
|
|
PORTAINER_KEY = "REDACTED_PORTAINER_TOKEN" # pragma: allowlist secret
|
|
|
|
ENDPOINTS = {
|
|
"atlantis": 2,
|
|
"calypso": 443397,
|
|
"nuc": 443398,
|
|
"homelab": 443399,
|
|
"rpi5": 443395,
|
|
}
|
|
|
|
|
|
def portainer_api(method: str, path: str, data: dict | None = None,
|
|
url: str = PORTAINER_URL, key: str = PORTAINER_KEY) -> dict | list:
|
|
"""Make a Portainer API request."""
|
|
full_url = f"{url.rstrip('/')}/api/{path.lstrip('/')}"
|
|
body = json.dumps(data).encode() if data else None
|
|
req = urllib.request.Request(full_url, data=body, method=method, headers={
|
|
"X-API-Key": key,
|
|
"Content-Type": "application/json",
|
|
})
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read())
|
|
|
|
|
|
def list_containers(endpoint: str, all_containers: bool = True) -> list[dict]:
|
|
"""List containers on an endpoint."""
|
|
eid = ENDPOINTS.get(endpoint, endpoint)
|
|
params = "all=true" if all_containers else "all=false"
|
|
return portainer_api("GET", f"endpoints/{eid}/docker/containers/json?{params}")
|
|
|
|
|
|
def get_container_logs(endpoint: str, container_id: str, tail: int = 100) -> str:
|
|
"""Get container logs."""
|
|
eid = ENDPOINTS.get(endpoint, endpoint)
|
|
url = f"{PORTAINER_URL}/api/endpoints/{eid}/docker/containers/{container_id}/logs?stdout=true&stderr=true&tail={tail}"
|
|
req = urllib.request.Request(url, headers={"X-API-Key": PORTAINER_KEY})
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
raw = resp.read()
|
|
# Strip Docker log prefix bytes (8-byte header per line)
|
|
lines = []
|
|
for line in raw.split(b"\n"):
|
|
if len(line) > 8:
|
|
lines.append(line[8:].decode("utf-8", errors="replace"))
|
|
return "\n".join(lines)
|
|
|
|
|
|
def restart_container(endpoint: str, container_id: str) -> bool:
|
|
"""Restart a container. Returns True on success."""
|
|
eid = ENDPOINTS.get(endpoint, endpoint)
|
|
try:
|
|
portainer_api("POST", f"endpoints/{eid}/docker/containers/{container_id}/restart")
|
|
return True
|
|
except urllib.error.HTTPError as e:
|
|
log.error("Restart failed for %s: %s", container_id, e)
|
|
return False
|
|
|
|
|
|
def inspect_container(endpoint: str, container_id: str) -> dict:
|
|
"""Inspect a container for full config."""
|
|
eid = ENDPOINTS.get(endpoint, endpoint)
|
|
return portainer_api("GET", f"endpoints/{eid}/docker/containers/{container_id}/json")
|