Enhance DockerComposeManager with detailed service, volume, and network configurations using Pydantic models

pyproject.toml

@@ -4,7 +4,12 @@ version = "0.1.0"
 description = "A simple Python package for managing Docker Compose files"
 readme = "README.md"
 requires-python = ">=3.13"
-dependencies = ["pytest>=8.4.0", "pyyaml>=6.0.2"]
+dependencies = [
+    "pydantic>=2.11.7",
+    "pytest>=8.4.0",
+    "pyyaml>=6.0.2",
+    "ruff>=0.12.0",
+]
 
 [build-system]
 requires = ["hatchling"]

src/compose/__init__.py

@@ -14,13 +14,59 @@ from pathlib import Path
 from typing import TYPE_CHECKING, Any, Self
 
 import yaml
+from pydantic import BaseModel, ValidationError
+
+
+class ServiceConfig(BaseModel):
+    image: str = ""
+    ports: list[str] | None = None
+    environment: dict[str, str] | None = None
+    volumes: list[str] | None = None
+    networks: list[str] | None = None
+    command: str | list[str] | None = None
+    entrypoint: str | list[str] | None = None
+    build: dict[str, Any] | str | None = None
+    healthcheck: dict[str, Any] | None = None
+    restart: str | None = None
+    labels: dict[str, str] | list[str] | None = None
+    depends_on: list[str] | None = None
+    configs: list[dict[str, Any]] | None = None
+    secrets: list[dict[str, Any]] | None = None
+    deploy: dict[str, Any] | None = None
+    resources: dict[str, Any] | None = None
+
+    model_config = {"extra": "allow"}
+
+
+class VolumeConfig(BaseModel):
+    # Add more fields as needed for Docker Compose volumes
+    driver: str | None = None
+    driver_opts: dict[str, Any] | None = None
+    external: bool | dict[str, Any] | None = None
+    labels: dict[str, str] | list[str] | None = None
+    name: str | None = None
+
+    model_config = {"extra": "allow"}
+
+
+class NetworkConfig(BaseModel):
+    # Add more fields as needed for Docker Compose networks
+    driver: str | None = None
+    driver_opts: dict[str, Any] | None = None
+    attachable: bool | None = None
+    external: bool | dict[str, Any] | None = None
+    labels: dict[str, str] | list[str] | None = None
+    name: str | None = None
+
+    model_config = {"extra": "allow"}
+
 
 if TYPE_CHECKING:
     from types import TracebackType
 
 
 class DockerComposeManager:
-    def add_volume(self, name: str, config: dict[str, Any] | None = None) -> DockerComposeManager:
+    def add_volume(self, name: str, config: VolumeConfig | dict[str, Any] | None = None) -> DockerComposeManager:
         """Add a top-level volume definition.
 
         Returns:
@@ -28,7 +74,12 @@ class DockerComposeManager:
         """
         if "volumes" not in self._data or not isinstance(self._data["volumes"], dict):
             self._data["volumes"] = {}
-        self._data["volumes"][name] = config or {}
+        if config is None:
+            self._data["volumes"][name] = {}
+        else:
+            if isinstance(config, dict):
+                config = VolumeConfig(**config)
+            self._data["volumes"][name] = config.model_dump(exclude_none=True)
         self._dirty = True
         return self
 
@@ -43,7 +94,7 @@ class DockerComposeManager:
             self._dirty = True
         return self
 
-    def add_network(self, name: str, config: dict[str, Any] | None = None) -> DockerComposeManager:
+    def add_network(self, name: str, config: NetworkConfig | dict[str, Any] | None = None) -> DockerComposeManager:
         """Add a top-level network definition.
 
         Returns:
@@ -51,7 +102,12 @@ class DockerComposeManager:
         """
         if "networks" not in self._data or not isinstance(self._data["networks"], dict):
             self._data["networks"] = {}
-        self._data["networks"][name] = config or {}
+        if config is None:
+            self._data["networks"][name] = {}
+        else:
+            if isinstance(config, dict):
+                config = NetworkConfig(**config)
+            self._data["networks"][name] = config.model_dump(exclude_none=True)
         self._dirty = True
         return self
 
@@ -93,6 +149,8 @@ class DockerComposeManager:
     def create_service(
         self,
         name: str,
+        *,
+        config: ServiceConfig | dict[str, Any] | None = None,
         image: str = "",
         ports: list[str] | None = None,
         environment: dict[str, str] | None = None,
@@ -116,42 +174,39 @@ class DockerComposeManager:
         Returns:
             DockerComposeManager: self (for chaining)
 
+        Raises:
+            ValueError: If the service config is invalid.
         """
         services = self._data["services"]
-        service: dict[str, Any] = {}
+        if config is not None:
-        if image:
+            if isinstance(config, dict):
-            service["image"] = image
+                config = ServiceConfig(**config)
-        if ports is not None:
+            service = config.model_dump(exclude_none=True)
-            service["ports"] = ports
+            service.update(kwargs)
-        if environment is not None:
+        else:
-            service["environment"] = environment
+            try:
-        if volumes is not None:
+                service = ServiceConfig(
-            service["volumes"] = volumes
+                    image=image,
-        if networks is not None:
+                    ports=ports,
-            service["networks"] = networks
+                    environment=environment,
-        if command is not None:
+                    volumes=volumes,
-            service["command"] = command
+                    networks=networks,
-        if entrypoint is not None:
+                    command=command,
-            service["entrypoint"] = entrypoint
+                    entrypoint=entrypoint,
-        if build is not None:
+                    build=build,
-            service["build"] = build
+                    healthcheck=healthcheck,
-        if healthcheck is not None:
+                    restart=restart,
-            service["healthcheck"] = healthcheck
+                    labels=labels,
-        if restart is not None:
+                    depends_on=depends_on,
-            service["restart"] = restart
+                    configs=configs,
-        if labels is not None:
+                    secrets=secrets,
-            service["labels"] = labels
+                    deploy=deploy,
-        if depends_on is not None:
+                    resources=resources,
-            service["depends_on"] = depends_on
+                    **kwargs,
-        if configs is not None:
+                ).model_dump(exclude_none=True)
-            service["configs"] = configs
+            except ValidationError as e:
-        if secrets is not None:
+                msg = f"Invalid service config: {e}"
-            service["secrets"] = secrets
+                raise ValueError(msg) from e
-        if deploy is not None:
-            service["deploy"] = deploy
-        if resources is not None:
-            service["resources"] = resources
-        service.update(kwargs)
         services[name] = service
         self._dirty = True
         return self