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

example/basic_example.py

@@ -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)

src/compose/__init__.py

@@ -1,6 +1,3 @@
-# Copyright (c) 2023, the Compose project contributors.
-# SPDX-License-Identifier: GPL-3.0-or-later
-
 """Docker Compose YAML file generator.
 
 This package provides utilities for programmatically creating and managing Docker Compose
@@ -29,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
@@ -51,6 +48,9 @@ class VolumeConfig(BaseModel):
     labels: dict[str, str] | list[str] | None = None
     name: str | None = None
 
+    # Allow extra fields for flexibility and to support arbitrary Docker Compose extensions.
+    model_config = {"extra": "allow"}
+
 
 class NetworkConfig(BaseModel):
     """Represents a network configuration for a Docker Compose file."""
@@ -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.

tests/test_compose_manager.py

@@ -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: