Compare commits
2 Commits
f0ee0bcac6
...
2996016aee
Author | SHA1 | Date | |
---|---|---|---|
2996016aee | |||
2e000017e4 |
12
.vscode/settings.json
vendored
Normal file
12
.vscode/settings.json
vendored
Normal file
@ -0,0 +1,12 @@
|
||||
{
|
||||
"cSpell.words": [
|
||||
"docstrings",
|
||||
"healthcheck",
|
||||
"isort",
|
||||
"pycodestyle",
|
||||
"pydantic",
|
||||
"pydocstyle",
|
||||
"pytest",
|
||||
"pyyaml"
|
||||
]
|
||||
}
|
@ -1,7 +1,14 @@
|
||||
"""Example usage of DockerComposeManager to generate a docker-compose.yaml file."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from compose import DockerComposeManager
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
logger: logging.Logger = logging.getLogger("docker-compose-example")
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Path to the compose file to generate
|
||||
compose_path = "docker-compose.yaml"
|
||||
@ -24,7 +31,7 @@ if __name__ == "__main__":
|
||||
environment={
|
||||
"POSTGRES_USER": "user",
|
||||
"POSTGRES_PASSWORD": "password",
|
||||
"POSTGRES_DB": "exampledb",
|
||||
"POSTGRES_DB": "example_db",
|
||||
},
|
||||
ports=["5432:5432"],
|
||||
volumes=["db_data:/var/lib/postgresql/data"],
|
||||
@ -33,4 +40,4 @@ if __name__ == "__main__":
|
||||
# Save the compose file
|
||||
manager.save()
|
||||
|
||||
print(f"docker-compose.yaml generated at {compose_path}")
|
||||
logger.info("docker-compose.yaml generated at %s", compose_path)
|
||||
|
@ -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"]
|
||||
|
@ -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
|
||||
|
Reference in New Issue
Block a user