ez_docker/plugins/module_utils/common.py
2024-12-11 18:42:59 +01:00

153 lines
4 KiB
Python

# Copyright: (c) 2024, Luca Bilke <luca@bil.ke>
# MIT License (see LICENSE)
"""Common module utilities."""
from __future__ import annotations
import copy
from dataclasses import asdict, dataclass, field, replace
from pathlib import Path
from typing import Any, Callable
import yaml
from ansible.module_utils.basic import AnsibleModule
PROJECTS_DIR = "/var/lib/ez_compose"
BASE_ARGS = {
"project_name": {
"type": "str",
"required": True,
},
"name": {
"type": "str",
"required": True,
},
}
@dataclass(frozen=True)
class Result:
"""Module result object."""
changed: bool = False
diff: dict[str, Any] = field(default_factory=dict)
# @dataclass(frozen=True)
# class Settings:
# projects_dir: str = "/usr/local/share/ez_compose/"
@dataclass(frozen=True)
class State:
"""Execution state object."""
module: AnsibleModule
result: Result
compose_filepath: str
before: dict[str, Any]
after: dict[str, Any]
def recursive_update(
default: dict[Any, Any],
update: dict[Any, Any],
) -> dict[Any, Any]:
"""Recursively update a dictionary."""
for key in update: # noqa: PLC0206
if isinstance(update[key], dict) and isinstance(default.get(key), dict):
default[key] = recursive_update(default[key], update[key])
elif isinstance(update[key], list) and isinstance(default.get(key), list):
default_set = set(default[key])
custom_set = set(update[key])
default[key] = list(default_set.union(custom_set))
else:
default[key] = update[key]
return default
def get_state(module: AnsibleModule) -> State:
"""Create a new state object, loading the compose file into "before" if it exists."""
compose_filepath = (
f"{PROJECTS_DIR}/{module.params['project_name']}/docker-compose.yml"
)
try:
with Path(compose_filepath).open("r") as fp:
before = yaml.safe_load(fp)
except FileNotFoundError:
before: dict[str, Any] = {}
return State(
module=module,
result=Result(),
compose_filepath=compose_filepath,
before=before,
after={},
)
def update_project(state: State) -> State:
"""Ensure that networks/volumes that exist in services also exist in the project."""
project = copy.deepcopy(state.after)
project_services = project.get("services", {})
project_networks = project.get("networks", {})
project_volumes = project.get("volumes", {})
for service in project_services:
if service_volumes := service.get("volumes"):
service_volume_names = [x["source"] for x in service_volumes]
project_volumes.update(
{
service_volume_name: None
for service_volume_name in service_volume_names
if service_volume_name not in project_volumes
},
)
if service_network_names := service.get("networks").keys():
project_networks.update(
{
service_network_name: None
for service_network_name in service_network_names
if service_network_name not in project_networks
},
)
return replace(state, after=project)
def write_compose(state: State) -> State:
"""Write the compose file to disk."""
file = state.compose_filepath
with Path(file).open(mode="w") as stream:
yaml.dump(state.after, stream)
return state
def run_module(
args: dict[str, Any] | None = None,
execute: Callable[[State], State] = lambda _: _,
) -> None:
"""Handle module setup and teardown."""
module = AnsibleModule(
argument_spec={**BASE_ARGS, **(args or {})},
supports_check_mode=True,
)
state = get_state(module)
state = execute(state)
if not state.module.check_mode:
write_compose(state)
state.module.exit_json(**asdict(state.result)) # type: ignore[reportUnkownMemberType]