Developer
f6003b75ca
- Create services/docker/compose.py — compose file generation and commands - Create services/docker/container.py — container lifecycle and queries - Create services/docker/config_staging.py — config folder file writing - Create services/docker/tunnel.py — Cloudflare tunnel management - Create services/docker/__init__.py — barrel exports - Delete services/docker.py (replaced by package) - All imports in api/tool_instances.py remain functional Quality gates: Python syntax check (pass), imports verified Refs: repo-restructure Task 3.3
122 lines
3.1 KiB
Python
122 lines
3.1 KiB
Python
"""Docker container lifecycle and query operations."""
|
|
|
|
import socket
|
|
import subprocess
|
|
|
|
|
|
def get_container_id(instance_name: str) -> str | None:
|
|
"""Get the container ID for a compose service.
|
|
|
|
Args:
|
|
instance_name: The service name in compose
|
|
|
|
Returns:
|
|
Container ID or None if not found
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "ps", "-q", "--filter", f"name={instance_name}"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
return result.stdout.strip().split("\n")[0]
|
|
return None
|
|
|
|
|
|
def get_container_name(instance_name: str) -> str | None:
|
|
"""Get the full container name for a compose service.
|
|
|
|
Args:
|
|
instance_name: The service name in compose
|
|
|
|
Returns:
|
|
Container name or None if not found
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "ps", "--format", "{{.Names}}", "--filter", f"name={instance_name}"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
return result.stdout.strip().split("\n")[0]
|
|
return None
|
|
|
|
|
|
def connect_container_to_network(container_name: str, network_name: str = "backend") -> bool:
|
|
"""Connect a Docker container to an existing network.
|
|
|
|
Args:
|
|
container_name: Name or ID of the container
|
|
network_name: Name of the Docker network (default: backend)
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "network", "connect", network_name, container_name],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return result.returncode == 0
|
|
|
|
|
|
def get_container_status(container_id: str) -> str:
|
|
"""Get the status of a Docker container.
|
|
|
|
Args:
|
|
container_id: Docker container ID
|
|
|
|
Returns:
|
|
Container status string (running, exited, etc.)
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "inspect", "-f", "{{.State.Status}}", container_id],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
return result.stdout.strip()
|
|
return "unknown"
|
|
|
|
|
|
def get_container_logs(container_id: str, tail: int = 100) -> str:
|
|
"""Get the logs of a Docker container.
|
|
|
|
Args:
|
|
container_id: Docker container ID
|
|
tail: Number of lines to return
|
|
|
|
Returns:
|
|
Container logs
|
|
"""
|
|
result = subprocess.run(
|
|
["docker", "logs", "--tail", str(tail), container_id],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode == 0:
|
|
return result.stdout
|
|
return f"Failed to get logs: {result.stderr}"
|
|
|
|
|
|
def find_free_port(start: int = 10000, end: int = 20000) -> int:
|
|
"""Find a free TCP port in the given range.
|
|
|
|
Args:
|
|
start: Start of port range
|
|
end: End of port range
|
|
|
|
Returns:
|
|
Free port number
|
|
"""
|
|
for port in range(start, end):
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
if s.connect_ex(("localhost", port)) != 0:
|
|
return port
|
|
|
|
raise RuntimeError(f"No free port found in range {start}-{end}")
|