diff --git a/pyproject.toml b/pyproject.toml index d4cfb41..3f6b375 100644 --- a/pyproject.toml +++ b/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"] diff --git a/src/compose/__init__.py b/src/compose/__init__.py index acbf5e0..b434e4d 100644 --- a/src/compose/__init__.py +++ b/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 image: - service["image"] = image - if ports is not None: - service["ports"] = ports - if environment is not None: - service["environment"] = environment - if volumes is not None: - service["volumes"] = volumes - if networks is not None: - service["networks"] = networks - if command is not None: - service["command"] = command - if entrypoint is not None: - service["entrypoint"] = entrypoint - if build is not None: - service["build"] = build - if healthcheck is not None: - service["healthcheck"] = healthcheck - if restart is not None: - service["restart"] = restart - if labels is not None: - service["labels"] = labels - if depends_on is not None: - service["depends_on"] = depends_on - if configs is not None: - service["configs"] = configs - if secrets is not None: - service["secrets"] = secrets - if deploy is not None: - service["deploy"] = deploy - if resources is not None: - service["resources"] = resources - service.update(kwargs) + if config is not None: + if isinstance(config, dict): + config = ServiceConfig(**config) + service = config.model_dump(exclude_none=True) + service.update(kwargs) + else: + try: + service = ServiceConfig( + image=image, + ports=ports, + environment=environment, + volumes=volumes, + networks=networks, + command=command, + entrypoint=entrypoint, + build=build, + healthcheck=healthcheck, + restart=restart, + labels=labels, + depends_on=depends_on, + configs=configs, + secrets=secrets, + deploy=deploy, + resources=resources, + **kwargs, + ).model_dump(exclude_none=True) + except ValidationError as e: + msg = f"Invalid service config: {e}" + raise ValueError(msg) from e services[name] = service self._dirty = True return self