ez_docker/plugins/module_utils/common.py
2025-01-22 18:48:59 +01:00

165 lines
4.7 KiB
Python

# Copyright: (c) 2025, Luca Bilke <luca@bil.ke>
# MIT License (see LICENSE)
from __future__ import annotations
import copy
from dataclasses import replace
from pathlib import Path
from typing import TYPE_CHECKING, Any, TypeVar, cast
import yaml
from ansible_collections.snailed.ez_docker.plugins.module_utils.models import Result, State
if TYPE_CHECKING:
from ansible.module_utils.basic import AnsibleModule # pyright: ignore[reportMissingTypeStubs]
def recursive_update(
default: dict[str, Any],
update: dict[str, Any],
) -> dict[str, Any]:
default = copy.deepcopy(default)
for k, v in update.items():
if isinstance(v, dict):
v = cast(dict[str, Any], v)
if v.get("_ezd_no_defaults"):
v.pop("_ezd_no_defaults")
default[k] = v
else:
default[k] = recursive_update(default.get(k, {}), v)
elif isinstance(v, list):
v = cast(list[Any], v)
old = cast(list[Any], (default.get(k, [])))
if "_ezd_no_defaults" in v:
v.remove("_ezd_no_defaults")
default[k] = v
else:
old.extend(v)
default[k] = old
else:
default[k] = v
return default
def get_state(module: AnsibleModule) -> State:
"""Create a new state object, loading the compose file into "before" if it exists."""
# def clean_params[T](obj: T) -> T:
T = TypeVar("T")
def clean_params(obj: T) -> T:
obj = copy.deepcopy(obj)
if isinstance(obj, dict):
return {key: clean_params(value) for key, value in obj.items() if value is not None} # pyright: ignore[reportUnknownArgumentType, reportUnknownVariableType, reportReturnType]
if isinstance(obj, list):
return [clean_params(item) for item in obj if item is not None] # pyright: ignore[reportUnknownArgumentType, reportUnknownVariableType, reportReturnType]
return obj
compose_filepath = f"{module.params['project_dir']}/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,
params=clean_params(module.params),
result=Result(),
compose_filepath=compose_filepath,
before=before,
after={
"name": module.params["name"],
"services": {},
"networks": {},
"volumes": {},
},
)
def update_project(state: State) -> State:
"""Properly configure top level volume/network elements."""
project = copy.deepcopy(state.after)
services: dict[str, Any] = project.get("services", {})
settings: dict[str, Any] = state.params.get("settings", {})
volume_sources = [
vol["source"] for service in services.values() for vol in service.get("volumes", [])
if vol.get("type") == "volume"
]
network_names = [
network for service in services.values() for network in service.get("networks", {})
]
project["volumes"] = {source: None for source in volume_sources} | {
source: {"external": True}
for source in volume_sources
if source in settings.get("external_volumes", [])
}
project["networks"] = {
network: ({"external": True} if network in settings.get("external_networks", []) else None)
for network in network_names
}
return replace(state, after=project)
def set_result(state: State) -> State: # noqa: C901
def _changed(before: Any, after: Any) -> bool: # noqa: ANN401, C901, PLR0911
if type(before) is not type(after):
return True
if isinstance(before, dict):
before = cast(dict[str, Any], before)
if len(before) != len(after):
return True
for key in before:
if key not in after:
return True
if _changed(before[key], after[key]):
return True
return False
if isinstance(before, list):
before = cast(list[Any], before)
if len(before) != len(after):
return True
for index, _ in enumerate(before):
if _changed(before[index], after[index]):
return True
return before != after
result = Result(
_changed(state.before, state.after),
{
"before": state.before,
"after": state.after,
},
)
return replace(state, result=result)
def write_compose(state: State) -> State:
file = state.compose_filepath
with Path(file).open(mode="w") as stream:
yaml.dump(state.after, stream)
return state