TheLovinator/compose
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor DockerComposeManager usage in basic_example and enhance error handling for service, network, config, and secret management

Browse Source
This commit is contained in:
Joakim Hellsén
2025-06-18 05:05:29 +02:00
parent c8e262a736
commit 2557aacd5d
3 changed files with 162 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
example/basic_example.py
View File

@ -6,38 +6,53 @@ import logging
from compose import DockerComposeManager
logging.basicConfig(level=logging.DEBUG)
logging.basicConfig(level=logging.INFO)
logger: logging.Logger = logging.getLogger("docker-compose-example")
if __name__ == "__main__":
# Path to the compose file to generate
compose_path = "docker-compose.yaml"
# Create a DockerComposeManager instance
manager = DockerComposeManager(compose_path)
# Using DockerComposeManager as a context manager
with DockerComposeManager(compose_path) as manager:
# Add top-level networks, volumes, configs, and secrets
manager.add_network("my_network")
manager.add_volume("db_data")
manager.add_config("my_config", config={"file": "./config.json"})
manager.add_secret("my_secret", config={"file": "./secret.txt"})
# Add a simple web service
manager.create_service(
name="web",
image="nginx:alpine",
ports=["8080:80"],
environment={"NGINX_HOST": "localhost"},
)
# Add a simple web service
manager.create_service(
name="web",
image="nginx:alpine",
ports=["8080:80"],
environment={"NGINX_HOST": "localhost"},
networks=["my_network"],
)
# Add a database service
manager.create_service(
name="db",
image="postgres:15-alpine",
environment={
"POSTGRES_USER": "user",
"POSTGRES_PASSWORD": "password",
"POSTGRES_DB": "example_db",
},
ports=["5432:5432"],
volumes=["db_data:/var/lib/postgresql/data"],
)
# Add a database service that depends on the web service
manager.create_service(
name="db",
image="postgres:15-alpine",
environment={
"POSTGRES_USER": "user",
"POSTGRES_PASSWORD": "password",
"POSTGRES_DB": "example_db",
},
ports=["5432:5432"],
volumes=["db_data:/var/lib/postgresql/data"],
networks=["my_network"],
depends_on={"web": {"condition": "service_started"}},
)
# Save the compose file
manager.save()
# Modify the web service
manager.modify_service("web", ports=["8081:80"])
# Add another service and then remove it
manager.create_service("temp_service", image="alpine:latest")
manager.remove_service("temp_service")
# Remove a network
manager.remove_network("my_network")
logger.info("docker-compose.yaml generated at %s", compose_path)

133
src/compose/__init__.py
View File

@ -26,7 +26,7 @@ class ServiceConfig(BaseModel):
healthcheck: dict[str, Any] | None = None
restart: str | None = None
labels: dict[str, str] | list[str] | None = None
depends_on: list[str] | None = None
depends_on: list[str] | dict[str, dict[str, str]] | None = None
configs: list[dict[str, Any]] | None = None
secrets: list[dict[str, Any]] | None = None
deploy: dict[str, Any] | None = None
@ -66,6 +66,28 @@ class NetworkConfig(BaseModel):
model_config = {"extra": "allow"}
class ConfigConfig(BaseModel):
"""Represents a config configuration for a Docker Compose file."""
file: str | None = None
external: bool | None = None
name: str | None = None
# Allow extra fields for flexibility and to support arbitrary Docker Compose extensions.
model_config = {"extra": "allow"}
class SecretConfig(BaseModel):
"""Represents a secret configuration for a Docker Compose file."""
file: str | None = None
external: bool | None = None
name: str | None = None
# Allow extra fields for flexibility and to support arbitrary Docker Compose extensions.
model_config = {"extra": "allow"}
if TYPE_CHECKING:
from types import TracebackType
@ -98,10 +120,17 @@ class DockerComposeManager:
Returns:
DockerComposeManager: self (for chaining)
Raises:
KeyError: If the volume does not exist.
"""
if "volumes" in self._data and name in self._data["volumes"]:
del self._data["volumes"][name]
self._dirty = True
if "volumes" not in self._data or name not in self._data["volumes"]:
msg = f"Volume '{name}' not found."
raise KeyError(msg)
del self._data["volumes"][name]
if not self._data["volumes"]:
del self._data["volumes"]
self._dirty = True
return self
def add_network(self, name: str, config: NetworkConfig | dict[str, Any] | None = None) -> DockerComposeManager:
@ -126,10 +155,87 @@ class DockerComposeManager:
Returns:
DockerComposeManager: self (for chaining)
Raises:
KeyError: If the network does not exist.
"""
if "networks" in self._data and name in self._data["networks"]:
del self._data["networks"][name]
self._dirty = True
if "networks" not in self._data or name not in self._data["networks"]:
msg = f"Network '{name}' not found."
raise KeyError(msg)
del self._data["networks"][name]
if not self._data["networks"]:
del self._data["networks"]
self._dirty = True
return self
def add_config(self, name: str, config: ConfigConfig | dict[str, Any] | None = None) -> DockerComposeManager:
"""Add a top-level config definition.
Returns:
DockerComposeManager: self (for chaining)
"""
if "configs" not in self._data or not isinstance(self._data["configs"], dict):
self._data["configs"] = {}
if config is None:
self._data["configs"][name] = {}
else:
if isinstance(config, dict):
config = ConfigConfig(**config)
self._data["configs"][name] = config.model_dump(exclude_none=True)
self._dirty = True
return self
def remove_config(self, name: str) -> DockerComposeManager:
"""Remove a top-level config definition.
Returns:
DockerComposeManager: self (for chaining)
Raises:
KeyError: If the config does not exist.
"""
if "configs" not in self._data or name not in self._data["configs"]:
msg = f"Config '{name}' not found."
raise KeyError(msg)
del self._data["configs"][name]
if not self._data["configs"]:
del self._data["configs"]
self._dirty = True
return self
def add_secret(self, name: str, config: SecretConfig | dict[str, Any] | None = None) -> DockerComposeManager:
"""Add a top-level secret definition.
Returns:
DockerComposeManager: self (for chaining)
"""
if "secrets" not in self._data or not isinstance(self._data["secrets"], dict):
self._data["secrets"] = {}
if config is None:
self._data["secrets"][name] = {}
else:
if isinstance(config, dict):
config = SecretConfig(**config)
self._data["secrets"][name] = config.model_dump(exclude_none=True)
self._dirty = True
return self
def remove_secret(self, name: str) -> DockerComposeManager:
"""Remove a top-level secret definition.
Returns:
DockerComposeManager: self (for chaining)
Raises:
KeyError: If the secret does not exist.
"""
if "secrets" not in self._data or name not in self._data["secrets"]:
msg = f"Secret '{name}' not found."
raise KeyError(msg)
del self._data["secrets"][name]
if not self._data["secrets"]:
del self._data["secrets"]
self._dirty = True
return self
def __init__(self, path: str, version: str = "3.8") -> None:
@ -167,7 +273,7 @@ class DockerComposeManager:
healthcheck: dict[str, Any] | None = None,
restart: str | None = None,
labels: dict[str, str] | list[str] | None = None,
depends_on: list[str] | None = None,
depends_on: list[str] | dict[str, dict[str, str]] | None = None,
configs: list[dict[str, Any]] | None = None,
secrets: list[dict[str, Any]] | None = None,
deploy: dict[str, Any] | None = None,
@ -244,11 +350,15 @@ class DockerComposeManager:
Returns:
DockerComposeManager: self (for chaining)
Raises:
KeyError: If the service does not exist.
"""
services: dict[str, dict[str, Any]] = self._data["services"]
if name in services:
del services[name]
self._dirty = True
if name not in services:
msg: str = f"Service '{name}' not found."
raise KeyError(msg)
del services[name]
self._dirty = True
return self
def save(self) -> None:
@ -256,7 +366,6 @@ class DockerComposeManager:
with Path(self.path).open("w", encoding="utf-8") as f:
yaml.dump(self._data, f, sort_keys=False, indent=2, default_flow_style=False)
self._dirty = False
self._dirty = False
def __enter__(self) -> Self:
"""Enter the context manager and return self.

8
tests/test_compose_manager.py
View File

@ -71,12 +71,8 @@ def test_modify_nonexistent_service(tmp_path: Path) -> None:
def test_remove_nonexistent_service(tmp_path: Path) -> None:
compose_file: Path = tmp_path / "docker-compose.yml"
manager = DockerComposeManager(str(compose_file))
# Should not raise
manager.remove_service("notfound").save()
with compose_file.open() as f:
data = yaml.safe_load(f)
assert "services" in data
assert data["services"] == {}
with pytest.raises(KeyError):
manager.remove_service("notfound")
def test_create_service_with_extra_kwargs(tmp_path: Path) -> None: