ez_docker/plugins/module_utils/service/common.py
2024-12-28 21:22:44 +01:00

84 lines
2.4 KiB
Python

# Copyright: (c) 2024, Luca Bilke <luca@bil.ke>
# MIT License (see LICENSE)
from __future__ import annotations
import copy
from dataclasses import replace
from typing import TYPE_CHECKING, Any, Callable
from ansible_collections.ssnailed.ez_compose.plugins.module_utils.common import (
recursive_update,
update_project,
)
if TYPE_CHECKING:
from ansible_collections.ssnailed.ez_compose.plugins.module_utils.common import (
State,
)
BASE_ARGS: dict[str, Any] = {
"name": {"type": "str", "required": True},
"image": {"type": "str", "required": True},
"overwrite": {"type": "dict"},
}
def apply_base(state: State, params: dict[str, Any]) -> State:
project_name: str = state.module.params["name"]
new: dict[str, Any] = {
"container_name": f"{project_name}_{params["name"]}",
"hostname": f"{project_name}_{params["name"]}",
"image": params["image"],
"restart": "unless-stopped",
"environment": {},
"labels": {},
"volumes": [],
"networks": {
"internal": None,
},
}
return update(state, params, new)
def apply_definition(state: State, params: dict[str, Any], definition: dict[str, Any]) -> State:
service_name: str = params["name"]
project = copy.deepcopy(state.after)
services: dict[str, Any] = project["services"]
service: dict[str, Any] = services[service_name]
_ = recursive_update(service, definition)
services.update({service_name: service})
return replace(state, after=project)
def apply_settings(state: State, params: dict[str, Any]) -> State:
update = (
state.module.params.get("settings", {}).get("service_defaults", {}).get(params["name"], {})
)
return update(state, params, update)
def update(state: State, params: dict[str, Any], update: dict[str, Any]) -> State:
service_name: str = params["name"]
project = copy.deepcopy(state.after)
_ = recursive_update(project["services"][service_name], update)
return replace(state, after=project)
def run_helper(
state: State,
params: dict[str, Any],
helper: Callable[[State, dict[str, Any]], State] = lambda x, _: x,
) -> State:
state = apply_base(state, params)
state = apply_settings(state, params)
state = helper(state, params)
state = apply_definition(state, params, params.get("overwrite", {}))
return update_project(state)