menzeljandClaude Opus 4.8
1931500c24
- Live remote-stack logs over a WebSocket proxied through the central app to
the agent (/ws/agent-logs/{agent}/{stack}); agent gains a WS log endpoint.
- Deploy to a remote host from the UI: host selector in the New Stack editor
and template dialog; templates instantiate onto an agent via the proxy.
- Network attach/detach: expandable inspect view per network with
connect/disconnect + container picker; GET /{id}/containers, POST connect/disconnect.
- Remove dead pages/Placeholder.tsx.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
138 lines
4.4 KiB
Python
138 lines
4.4 KiB
Python
"""Docker network management."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional
|
|
|
|
from docker.errors import APIError
|
|
from docker.types import IPAMConfig, IPAMPool
|
|
|
|
from docker_client import DockerError, get_client, safe_call
|
|
|
|
COMPOSE_PROJECT_LABEL = "com.docker.compose.project"
|
|
DEFAULT_NETWORKS = {"bridge", "host", "none"}
|
|
|
|
|
|
def _summary(net) -> dict:
|
|
attrs = net.attrs
|
|
ipam = (attrs.get("IPAM") or {}).get("Config") or []
|
|
subnet = ipam[0].get("Subnet") if ipam else None
|
|
gateway = ipam[0].get("Gateway") if ipam else None
|
|
containers = attrs.get("Containers") or {}
|
|
names = [c.get("Name", cid[:12]) for cid, c in containers.items()]
|
|
labels = attrs.get("Labels") or {}
|
|
name = attrs.get("Name", net.name)
|
|
return {
|
|
"id": net.id[:12],
|
|
"name": name,
|
|
"driver": attrs.get("Driver"),
|
|
"scope": attrs.get("Scope"),
|
|
"internal": attrs.get("Internal", False),
|
|
"attachable": attrs.get("Attachable", False),
|
|
"subnet": subnet,
|
|
"gateway": gateway,
|
|
"containers": names,
|
|
"in_use": bool(names),
|
|
"stack": labels.get(COMPOSE_PROJECT_LABEL),
|
|
"labels": labels,
|
|
"created": attrs.get("Created"),
|
|
"is_default": name in DEFAULT_NETWORKS,
|
|
}
|
|
|
|
|
|
def list_networks() -> list[dict]:
|
|
client = get_client()
|
|
nets = safe_call(client.networks.list)
|
|
# list() entries are lightweight; reload for Containers/IPAM detail.
|
|
out = []
|
|
for n in nets:
|
|
try:
|
|
n.reload()
|
|
except APIError:
|
|
pass
|
|
out.append(_summary(n))
|
|
return sorted(out, key=lambda x: (x["is_default"] is False, x["name"]))
|
|
|
|
|
|
def inspect_network(network_id: str) -> dict:
|
|
client = get_client()
|
|
net = safe_call(client.networks.get, network_id)
|
|
return _summary(net)
|
|
|
|
|
|
def create_network(spec: dict) -> dict:
|
|
name = (spec.get("name") or "").strip()
|
|
if not name:
|
|
raise DockerError("invalid_request", "Network name is required")
|
|
driver = spec.get("driver") or "bridge"
|
|
ipam = None
|
|
subnet = (spec.get("subnet") or "").strip()
|
|
gateway = (spec.get("gateway") or "").strip()
|
|
if subnet:
|
|
pool = IPAMPool(subnet=subnet, gateway=gateway or None)
|
|
ipam = IPAMConfig(pool_configs=[pool])
|
|
|
|
client = get_client()
|
|
net = safe_call(
|
|
client.networks.create,
|
|
name=name,
|
|
driver=driver,
|
|
internal=bool(spec.get("internal")),
|
|
attachable=bool(spec.get("attachable", True)),
|
|
ipam=ipam,
|
|
options=spec.get("options") or None,
|
|
labels=spec.get("labels") or None,
|
|
)
|
|
net.reload()
|
|
return _summary(net)
|
|
|
|
|
|
def connectable_containers(network_id: str) -> list[dict]:
|
|
"""All containers on the host, flagged whether already on this network."""
|
|
client = get_client()
|
|
net = safe_call(client.networks.get, network_id)
|
|
net.reload()
|
|
connected = set((net.attrs.get("Containers") or {}).keys())
|
|
out = []
|
|
for c in safe_call(client.containers.list, all=True):
|
|
labels = c.labels or {}
|
|
out.append(
|
|
{
|
|
"id": c.id[:12],
|
|
"name": c.name,
|
|
"state": c.status,
|
|
"stack": labels.get(COMPOSE_PROJECT_LABEL),
|
|
"connected": c.id in connected,
|
|
}
|
|
)
|
|
return sorted(out, key=lambda x: x["name"])
|
|
|
|
|
|
def connect_container(network_id: str, container: str, aliases: Optional[list[str]] = None) -> None:
|
|
if not (container or "").strip():
|
|
raise DockerError("invalid_request", "Container is required")
|
|
client = get_client()
|
|
net = safe_call(client.networks.get, network_id)
|
|
safe_call(net.connect, container, aliases=aliases or None)
|
|
|
|
|
|
def disconnect_container(network_id: str, container: str, force: bool = False) -> None:
|
|
if not (container or "").strip():
|
|
raise DockerError("invalid_request", "Container is required")
|
|
client = get_client()
|
|
net = safe_call(client.networks.get, network_id)
|
|
safe_call(net.disconnect, container, force=force)
|
|
|
|
|
|
def delete_network(network_id: str) -> None:
|
|
client = get_client()
|
|
net = safe_call(client.networks.get, network_id)
|
|
name = (net.attrs or {}).get("Name", net.name)
|
|
if name in DEFAULT_NETWORKS:
|
|
raise DockerError("forbidden", f"Cannot delete the default '{name}' network")
|
|
safe_call(net.remove)
|
|
|
|
|
|
def prune_networks() -> dict:
|
|
client = get_client()
|
|
return safe_call(client.networks.prune)
|