menzeljandClaude Opus 4.8
f732cb080b
Self-hosted Docker Compose manager. - Backend: FastAPI + docker-py + SQLite (JWT auth, file-first stacks, lifecycle, live status, WebSocket logs, docker-run converter, audit log) - Frontend: React + Vite + Tailwind (login/setup, dashboard, stacks, stack detail, Monaco editor, dark/light theme) - Deployment: docker-compose.yml, Dockerfiles, nginx reverse proxy Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
51 lines
1.4 KiB
Python
51 lines
1.4 KiB
Python
"""Singleton docker-py client wrapper.
|
|
|
|
All Docker access goes through this module. The Docker socket is NEVER exposed
|
|
to the frontend.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import functools
|
|
from typing import Any
|
|
|
|
import docker
|
|
from docker.errors import APIError, DockerException, NotFound
|
|
|
|
from config import settings
|
|
|
|
|
|
class DockerError(Exception):
|
|
"""Structured error surfaced to API callers."""
|
|
|
|
def __init__(self, error: str, detail: str = ""):
|
|
self.error = error
|
|
self.detail = detail
|
|
super().__init__(f"{error}: {detail}" if detail else error)
|
|
|
|
|
|
@functools.lru_cache(maxsize=1)
|
|
def get_client() -> docker.DockerClient:
|
|
try:
|
|
base_url = f"unix://{settings.DOCKER_SOCKET}"
|
|
client = docker.DockerClient(base_url=base_url)
|
|
client.ping()
|
|
return client
|
|
except DockerException as exc: # pragma: no cover - environment dependent
|
|
raise DockerError("docker_unavailable", str(exc)) from exc
|
|
|
|
|
|
def reset_client() -> None:
|
|
get_client.cache_clear()
|
|
|
|
|
|
def safe_call(fn, *args, **kwargs) -> Any:
|
|
"""Run a docker-py call, converting exceptions into DockerError."""
|
|
try:
|
|
return fn(*args, **kwargs)
|
|
except NotFound as exc:
|
|
raise DockerError("not_found", str(exc)) from exc
|
|
except APIError as exc:
|
|
raise DockerError("docker_api_error", str(exc)) from exc
|
|
except DockerException as exc:
|
|
raise DockerError("docker_error", str(exc)) from exc
|