ansible-collections/community.docker
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.docker.git synced 2026-03-16 04:04:31 +00:00
Code Packages Projects Releases Wiki Activity

Add typing to Docker Swarm modules.

Browse Source
This commit is contained in:
Felix Fontein 2025-10-24 08:59:10 +02:00
parent 86031cd7f6
commit 13d39a36eb
15 changed files with 306 additions and 246 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
plugins/module_utils/_swarm.py
View File

@ -28,7 +28,6 @@ from ansible_collections.community.docker.plugins.module_utils._version import (
class AnsibleDockerSwarmClient(AnsibleDockerClient):
def get_swarm_node_id(self) -> str | None:
"""
Get the 'NodeID' of the Swarm node or 'None' if host is not in Swarm. It returns the NodeID
@ -281,7 +280,7 @@ class AnsibleDockerSwarmClient(AnsibleDockerClient):
def get_node_name_by_id(self, nodeid: str) -> str:
return self.get_node_inspect(nodeid)["Description"]["Hostname"]
def get_unlock_key(self) -> str | None:
def get_unlock_key(self) -> dict[str, t.Any] | None:
if self.docker_py_version < LooseVersion("2.7.0"):
return None
return super().get_unlock_key()

31
plugins/modules/docker_config.py
View File

@ -198,6 +198,7 @@ config_name:
import base64
import hashlib
import traceback
import typing as t
try:
@ -220,9 +221,7 @@ from ansible_collections.community.docker.plugins.module_utils._util import (
class ConfigManager(DockerBaseClass):
def __init__(self, client, results):
def __init__(self, client: AnsibleDockerClient, results: dict[str, t.Any]) -> None:
super().__init__()
self.client = client
@ -253,10 +252,10 @@ class ConfigManager(DockerBaseClass):
if self.rolling_versions:
self.version = 0
self.data_key = None
self.configs = []
self.data_key: str | None = None
self.configs: list[dict[str, t.Any]] = []
def __call__(self):
def __call__(self) -> None:
self.get_config()
if self.state == "present":
self.data_key = hashlib.sha224(self.data).hexdigest()
@ -265,7 +264,7 @@ class ConfigManager(DockerBaseClass):
elif self.state == "absent":
self.absent()
def get_version(self, config):
def get_version(self, config: dict[str, t.Any]) -> int:
try:
return int(
config.get("Spec", {}).get("Labels", {}).get("ansible_version", 0)
@ -273,14 +272,14 @@ class ConfigManager(DockerBaseClass):
except ValueError:
return 0
def remove_old_versions(self):
def remove_old_versions(self) -> None:
if not self.rolling_versions or self.versions_to_keep < 0:
return
if not self.check_mode:
while len(self.configs) > max(self.versions_to_keep, 1):
self.remove_config(self.configs.pop(0))
def get_config(self):
def get_config(self) -> None:
"""Find an existing config."""
try:
configs = self.client.configs(filters={"name": self.name})
@ -299,9 +298,9 @@ class ConfigManager(DockerBaseClass):
config for config in configs if config["Spec"]["Name"] == self.name
]
def create_config(self):
def create_config(self) -> str | None:
"""Create a new config"""
config_id = None
config_id: str | dict[str, t.Any] | None = None
# We ca not see the data after creation, so adding a label we can use for idempotency check
labels = {"ansible_key": self.data_key}
if self.rolling_versions:
@ -325,18 +324,18 @@ class ConfigManager(DockerBaseClass):
self.client.fail(f"Error creating config: {exc}")
if isinstance(config_id, dict):
config_id = config_id["ID"]
return config_id["ID"]
return config_id
def remove_config(self, config):
def remove_config(self, config: dict[str, t.Any]) -> None:
try:
if not self.check_mode:
self.client.remove_config(config["ID"])
except APIError as exc:
self.client.fail(f"Error removing config {config['Spec']['Name']}: {exc}")
def present(self):
def present(self) -> None:
"""Handles state == 'present', creating or updating the config"""
if self.configs:
config = self.configs[-1]
@ -378,7 +377,7 @@ class ConfigManager(DockerBaseClass):
self.results["config_id"] = self.create_config()
self.results["config_name"] = self.name
def absent(self):
def absent(self) -> None:
"""Handles state == 'absent', removing the config"""
if self.configs:
for config in self.configs:
@ -386,7 +385,7 @@ class ConfigManager(DockerBaseClass):
self.results["changed"] = True
def main():
def main() -> None:
argument_spec = {
"name": {"type": "str", "required": True},
"state": {

29
plugins/modules/docker_node.py
View File

@ -134,6 +134,7 @@ node:
"""
import traceback
import typing as t
try:
@ -157,18 +158,19 @@ from ansible_collections.community.docker.plugins.module_utils._util import (
class TaskParameters(DockerBaseClass):
def __init__(self, client):
hostname: str
def __init__(self, client: AnsibleDockerSwarmClient) -> None:
super().__init__()
# Spec
self.name = None
self.labels = None
self.labels_state = None
self.labels_to_remove = None
self.labels: dict[str, t.Any] | None = None
self.labels_state: t.Literal["merge", "replace"] = "merge"
self.labels_to_remove: list[str] | None = None
# Node
self.availability = None
self.role = None
self.availability: t.Literal["active", "pause", "drain"] | None = None
self.role: t.Literal["worker", "manager"] | None = None
for key, value in client.module.params.items():
setattr(self, key, value)
@ -177,9 +179,9 @@ class TaskParameters(DockerBaseClass):
class SwarmNodeManager(DockerBaseClass):
def __init__(self, client, results):
def __init__(
self, client: AnsibleDockerSwarmClient, results: dict[str, t.Any]
) -> None:
super().__init__()
self.client = client
@ -192,10 +194,9 @@ class SwarmNodeManager(DockerBaseClass):
self.node_update()
def node_update(self):
def node_update(self) -> None:
if not (self.client.check_if_swarm_node(node_id=self.parameters.hostname)):
self.client.fail("This node is not part of a swarm.")
return
if self.client.check_if_swarm_node_is_down():
self.client.fail("Can not update the node. The node is down.")
@ -206,7 +207,7 @@ class SwarmNodeManager(DockerBaseClass):
self.client.fail(f"Failed to get node information for {exc}")
changed = False
node_spec = {
node_spec: dict[str, t.Any] = {
"Availability": self.parameters.availability,
"Role": self.parameters.role,
"Labels": self.parameters.labels,
@ -277,7 +278,7 @@ class SwarmNodeManager(DockerBaseClass):
self.results["changed"] = changed
def main():
def main() -> None:
argument_spec = {
"hostname": {"type": "str", "required": True},
"labels": {"type": "dict"},

12
plugins/modules/docker_node_info.py
View File

@ -87,6 +87,7 @@ nodes:
"""
import traceback
import typing as t
from ansible_collections.community.docker.plugins.module_utils._common import (
RequestException,
@ -103,9 +104,8 @@ except ImportError:
pass
def get_node_facts(client):
results = []
def get_node_facts(client: AnsibleDockerSwarmClient) -> list[dict[str, t.Any]]:
results: list[dict[str, t.Any]] = []
if client.module.params["self"] is True:
self_node_id = client.get_swarm_node_id()
@ -114,8 +114,8 @@ def get_node_facts(client):
return results
if client.module.params["name"] is None:
node_info = client.get_all_nodes_inspect()
return node_info
node_info_list = client.get_all_nodes_inspect()
return node_info_list
nodes = client.module.params["name"]
if not isinstance(nodes, list):
@ -130,7 +130,7 @@ def get_node_facts(client):
return results
def main():
def main() -> None:
argument_spec = {
"name": {"type": "list", "elements": "str"},
"self": {"type": "bool", "default": False},

31
plugins/modules/docker_secret.py
View File

@ -190,6 +190,7 @@ secret_name:
import base64
import hashlib
import traceback
import typing as t
try:
@ -212,9 +213,7 @@ from ansible_collections.community.docker.plugins.module_utils._util import (
class SecretManager(DockerBaseClass):
def __init__(self, client, results):
def __init__(self, client: AnsibleDockerClient, results: dict[str, t.Any]) -> None:
super().__init__()
self.client = client
@ -244,10 +243,10 @@ class SecretManager(DockerBaseClass):
if self.rolling_versions:
self.version = 0
self.data_key = None
self.secrets = []
self.data_key: str | None = None
self.secrets: list[dict[str, t.Any]] = []
def __call__(self):
def __call__(self) -> None:
self.get_secret()
if self.state == "present":
self.data_key = hashlib.sha224(self.data).hexdigest()
@ -256,7 +255,7 @@ class SecretManager(DockerBaseClass):
elif self.state == "absent":
self.absent()
def get_version(self, secret):
def get_version(self, secret: dict[str, t.Any]) -> int:
try:
return int(
secret.get("Spec", {}).get("Labels", {}).get("ansible_version", 0)
@ -264,14 +263,14 @@ class SecretManager(DockerBaseClass):
except ValueError:
return 0
def remove_old_versions(self):
def remove_old_versions(self) -> None:
if not self.rolling_versions or self.versions_to_keep < 0:
return
if not self.check_mode:
while len(self.secrets) > max(self.versions_to_keep, 1):
self.remove_secret(self.secrets.pop(0))
def get_secret(self):
def get_secret(self) -> None:
"""Find an existing secret."""
try:
secrets = self.client.secrets(filters={"name": self.name})
@ -290,9 +289,9 @@ class SecretManager(DockerBaseClass):
secret for secret in secrets if secret["Spec"]["Name"] == self.name
]
def create_secret(self):
def create_secret(self) -> str | None:
"""Create a new secret"""
secret_id = None
secret_id: str | dict[str, t.Any] | None = None
# We cannot see the data after creation, so adding a label we can use for idempotency check
labels = {"ansible_key": self.data_key}
if self.rolling_versions:
@ -312,18 +311,18 @@ class SecretManager(DockerBaseClass):
self.client.fail(f"Error creating secret: {exc}")
if isinstance(secret_id, dict):
secret_id = secret_id["ID"]
return secret_id["ID"]
return secret_id
def remove_secret(self, secret):
def remove_secret(self, secret: dict[str, t.Any]) -> None:
try:
if not self.check_mode:
self.client.remove_secret(secret["ID"])
except APIError as exc:
self.client.fail(f"Error removing secret {secret['Spec']['Name']}: {exc}")
def present(self):
def present(self) -> None:
"""Handles state == 'present', creating or updating the secret"""
if self.secrets:
secret = self.secrets[-1]
@ -357,7 +356,7 @@ class SecretManager(DockerBaseClass):
self.results["secret_id"] = self.create_secret()
self.results["secret_name"] = self.name
def absent(self):
def absent(self) -> None:
"""Handles state == 'absent', removing the secret"""
if self.secrets:
for secret in self.secrets:
@ -365,7 +364,7 @@ class SecretManager(DockerBaseClass):
self.results["changed"] = True
def main():
def main() -> None:
argument_spec = {
"name": {"type": "str", "required": True},
"state": {

1
plugins/modules/docker_stack_task_info.py
View File

@ -84,7 +84,6 @@ EXAMPLES = r"""
import json
import traceback
import typing as t
from ansible.module_utils.common.text.converters import to_native

110
plugins/modules/docker_swarm.py
View File

@ -292,6 +292,7 @@ actions:
import json
import traceback
import typing as t
try:
@ -314,40 +315,40 @@ from ansible_collections.community.docker.plugins.module_utils._util import (
class TaskParameters(DockerBaseClass):
def __init__(self):
def __init__(self) -> None:
super().__init__()
self.advertise_addr = None
self.listen_addr = None
self.remote_addrs = None
self.join_token = None
self.data_path_addr = None
self.data_path_port = None
self.advertise_addr: str | None = None
self.listen_addr: str | None = None
self.remote_addrs: list[str] | None = None
self.join_token: str | None = None
self.data_path_addr: str | None = None
self.data_path_port: int | None = None
self.spec = None
# Spec
self.snapshot_interval = None
self.task_history_retention_limit = None
self.keep_old_snapshots = None
self.log_entries_for_slow_followers = None
self.heartbeat_tick = None
self.election_tick = None
self.dispatcher_heartbeat_period = None
self.node_cert_expiry = None
self.name = None
self.labels = None
self.snapshot_interval: int | None = None
self.task_history_retention_limit: int | None = None
self.keep_old_snapshots: int | None = None
self.log_entries_for_slow_followers: int | None = None
self.heartbeat_tick: int | None = None
self.election_tick: int | None = None
self.dispatcher_heartbeat_period: int | None = None
self.node_cert_expiry: int | None = None
self.name: str | None = None
self.labels: dict[str, t.Any] | None = None
self.log_driver = None
self.signing_ca_cert = None
self.signing_ca_key = None
self.ca_force_rotate = None
self.autolock_managers = None
self.rotate_worker_token = None
self.rotate_manager_token = None
self.default_addr_pool = None
self.subnet_size = None
self.signing_ca_cert: str | None = None
self.signing_ca_key: str | None = None
self.ca_force_rotate: int | None = None
self.autolock_managers: bool | None = None
self.rotate_worker_token: bool | None = None
self.rotate_manager_token: bool | None = None
self.default_addr_pool: list[str] | None = None
self.subnet_size: int | None = None
@staticmethod
def from_ansible_params(client):
def from_ansible_params(client: AnsibleDockerSwarmClient) -> TaskParameters:
result = TaskParameters()
for key, value in client.module.params.items():
if key in result.__dict__:
@ -356,7 +357,7 @@ class TaskParameters(DockerBaseClass):
result.update_parameters(client)
return result
def update_from_swarm_info(self, swarm_info):
def update_from_swarm_info(self, swarm_info: dict[str, t.Any]) -> None:
spec = swarm_info["Spec"]
ca_config = spec.get("CAConfig") or {}
@ -400,7 +401,7 @@ class TaskParameters(DockerBaseClass):
if "LogDriver" in spec["TaskDefaults"]:
self.log_driver = spec["TaskDefaults"]["LogDriver"]
def update_parameters(self, client):
def update_parameters(self, client: AnsibleDockerSwarmClient) -> None:
assign = {
"snapshot_interval": "snapshot_interval",
"task_history_retention_limit": "task_history_retention_limit",
@ -427,7 +428,12 @@ class TaskParameters(DockerBaseClass):
params[dest] = value
self.spec = client.create_swarm_spec(**params)
def compare_to_active(self, other, client, differences):
def compare_to_active(
self,
other: TaskParameters,
client: AnsibleDockerSwarmClient,
differences: DifferenceTracker,
) -> DifferenceTracker:
for k in self.__dict__:
if k in (
"advertise_addr",
@ -459,26 +465,28 @@ class TaskParameters(DockerBaseClass):
class SwarmManager(DockerBaseClass):
def __init__(self, client, results):
def __init__(
self, client: AnsibleDockerSwarmClient, results: dict[str, t.Any]
) -> None:
super().__init__()
self.client = client
self.results = results
self.check_mode = self.client.check_mode
self.swarm_info = {}
self.swarm_info: dict[str, t.Any] = {}
self.state = client.module.params["state"]
self.force = client.module.params["force"]
self.node_id = client.module.params["node_id"]
self.state: t.Literal["present", "join", "absent", "remove"] = (
client.module.params["state"]
)
self.force: bool = client.module.params["force"]
self.node_id: str | None = client.module.params["node_id"]
self.differences = DifferenceTracker()
self.parameters = TaskParameters.from_ansible_params(client)
self.created = False
def __call__(self):
def __call__(self) -> None:
choice_map = {
"present": self.init_swarm,
"join": self.join,
@ -486,14 +494,14 @@ class SwarmManager(DockerBaseClass):
"remove": self.remove,
}
choice_map.get(self.state)()
choice_map[self.state]()
if self.client.module._diff or self.parameters.debug:
diff = {}
diff["before"], diff["after"] = self.differences.get_before_after()
self.results["diff"] = diff
def inspect_swarm(self):
def inspect_swarm(self) -> None:
try:
data = self.client.inspect_swarm()
json_str = json.dumps(data, ensure_ascii=False)
@ -507,7 +515,7 @@ class SwarmManager(DockerBaseClass):
except APIError:
pass
def get_unlock_key(self):
def get_unlock_key(self) -> dict[str, t.Any]:
default = {"UnlockKey": None}
if not self.has_swarm_lock_changed():
return default
@ -516,18 +524,18 @@ class SwarmManager(DockerBaseClass):
except APIError:
return default
def has_swarm_lock_changed(self):
return self.parameters.autolock_managers and (
def has_swarm_lock_changed(self) -> bool:
return bool(self.parameters.autolock_managers) and (
self.created or self.differences.has_difference_for("autolock_managers")
)
def init_swarm(self):
def init_swarm(self) -> None:
if not self.force and self.client.check_if_swarm_manager():
self.__update_swarm()
return
if not self.check_mode:
init_arguments = {
init_arguments: dict[str, t.Any] = {
"advertise_addr": self.parameters.advertise_addr,
"listen_addr": self.parameters.listen_addr,
"force_new_cluster": self.force,
@ -562,7 +570,7 @@ class SwarmManager(DockerBaseClass):
"UnlockKey": self.swarm_info.get("UnlockKey"),
}
def __update_swarm(self):
def __update_swarm(self) -> None:
try:
self.inspect_swarm()
version = self.swarm_info["Version"]["Index"]
@ -587,13 +595,12 @@ class SwarmManager(DockerBaseClass):
)
except APIError as exc:
self.client.fail(f"Can not update a Swarm Cluster: {exc}")
return
self.inspect_swarm()
self.results["actions"].append("Swarm cluster updated")
self.results["changed"] = True
def join(self):
def join(self) -> None:
if self.client.check_if_swarm_node():
self.results["actions"].append("This node is already part of a swarm.")
return
@ -614,7 +621,7 @@ class SwarmManager(DockerBaseClass):
self.differences.add("joined", parameter=True, active=False)
self.results["changed"] = True
def leave(self):
def leave(self) -> None:
if not self.client.check_if_swarm_node():
self.results["actions"].append("This node is not part of a swarm.")
return
@ -627,7 +634,7 @@ class SwarmManager(DockerBaseClass):
self.differences.add("joined", parameter="absent", active="present")
self.results["changed"] = True
def remove(self):
def remove(self) -> None:
if not self.client.check_if_swarm_manager():
self.client.fail("This node is not a manager.")
@ -655,11 +662,12 @@ class SwarmManager(DockerBaseClass):
self.results["changed"] = True
def _detect_remove_operation(client):
def _detect_remove_operation(client: AnsibleDockerSwarmClient) -> bool:
return client.module.params["state"] == "remove"
def main():
def main() -> None:
# TODO: missing option log_driver?
argument_spec = {
"advertise_addr": {"type": "str"},
"data_path_addr": {"type": "str"},

34
plugins/modules/docker_swarm_info.py
View File

@ -186,6 +186,7 @@ tasks:
"""
import traceback
import typing as t
try:
@ -207,16 +208,20 @@ from ansible_collections.community.docker.plugins.module_utils._util import (
class DockerSwarmManager(DockerBaseClass):
def __init__(self, client, results):
def __init__(
self, client: AnsibleDockerSwarmClient, results: dict[str, t.Any]
) -> None:
super().__init__()
self.client = client
self.results = results
self.verbose_output = self.client.module.params["verbose_output"]
listed_objects = ["tasks", "services", "nodes"]
listed_objects: list[t.Literal["nodes", "tasks", "services"]] = [
"tasks",
"services",
"nodes",
]
self.client.fail_task_if_not_swarm_manager()
@ -235,15 +240,16 @@ class DockerSwarmManager(DockerBaseClass):
if self.client.module.params["unlock_key"]:
self.results["swarm_unlock_key"] = self.get_docker_swarm_unlock_key()
def get_docker_swarm_facts(self):
def get_docker_swarm_facts(self) -> dict[str, t.Any]:
try:
return self.client.inspect_swarm()
except APIError as exc:
self.client.fail(f"Error inspecting docker swarm: {exc}")
def get_docker_items_list(self, docker_object=None, filters=None):
items = None
items_list = []
def get_docker_items_list(
self, docker_object: t.Literal["nodes", "tasks", "services"], filters=None
) -> list[dict[str, t.Any]]:
items_list: list[dict[str, t.Any]] = []
try:
if docker_object == "nodes":
@ -252,6 +258,8 @@ class DockerSwarmManager(DockerBaseClass):
items = self.client.tasks(filters=filters)
elif docker_object == "services":
items = self.client.services(filters=filters)
else:
raise ValueError(f"Invalid docker_object {docker_object}")
except APIError as exc:
self.client.fail(
f"Error inspecting docker swarm for object '{docker_object}': {exc}"
@ -276,7 +284,7 @@ class DockerSwarmManager(DockerBaseClass):
return items_list
@staticmethod
def get_essential_facts_nodes(item):
def get_essential_facts_nodes(item: dict[str, t.Any]) -> dict[str, t.Any]:
object_essentials = {}
object_essentials["ID"] = item.get("ID")
@ -298,7 +306,7 @@ class DockerSwarmManager(DockerBaseClass):
return object_essentials
def get_essential_facts_tasks(self, item):
def get_essential_facts_tasks(self, item: dict[str, t.Any]) -> dict[str, t.Any]:
object_essentials = {}
object_essentials["ID"] = item["ID"]
@ -319,7 +327,7 @@ class DockerSwarmManager(DockerBaseClass):
return object_essentials
@staticmethod
def get_essential_facts_services(item):
def get_essential_facts_services(item: dict[str, t.Any]) -> dict[str, t.Any]:
object_essentials = {}
object_essentials["ID"] = item["ID"]
@ -343,12 +351,12 @@ class DockerSwarmManager(DockerBaseClass):
return object_essentials
def get_docker_swarm_unlock_key(self):
def get_docker_swarm_unlock_key(self) -> str | None:
unlock_key = self.client.get_unlock_key() or {}
return unlock_key.get("UnlockKey") or None
def main():
def main() -> None:
argument_spec = {
"nodes": {"type": "bool", "default": False},
"nodes_filters": {"type": "dict"},

291
plugins/modules/docker_swarm_service.py
View File

@ -853,6 +853,7 @@ EXAMPLES = r"""
import shlex
import time
import traceback
import typing as t
from ansible.module_utils.basic import human_to_bytes
from ansible.module_utils.common.text.converters import to_text
@ -891,7 +892,9 @@ except ImportError:
pass
def get_docker_environment(env, env_files):
def get_docker_environment(
env: str | dict[str, t.Any] | list[t.Any] | None, env_files: list[str] | None
) -> list[str] | None:
"""
Will return a list of "KEY=VALUE" items. Supplied env variable can
be either a list or a dictionary.
@ -899,7 +902,7 @@ def get_docker_environment(env, env_files):
If environment files are combined with explicit environment variables,
the explicit environment variables take precedence.
"""
env_dict = {}
env_dict: dict[str, str] = {}
if env_files:
for env_file in env_files:
parsed_env_file = parse_env_file(env_file)
@ -936,7 +939,9 @@ def get_docker_environment(env, env_files):
return sorted(env_list)
def get_docker_networks(networks, network_ids):
def get_docker_networks(
networks: list[str | dict[str, t.Any]] | None, network_ids: dict[str, str]
) -> list[dict[str, t.Any]] | None:
"""
Validate a list of network names or a list of network dictionaries.
Network names will be resolved to ids by using the network_ids mapping.
@ -945,6 +950,7 @@ def get_docker_networks(networks, network_ids):
return None
parsed_networks = []
for network in networks:
parsed_network: dict[str, t.Any]
if isinstance(network, str):
parsed_network = {"name": network}
elif isinstance(network, dict):
@ -988,7 +994,7 @@ def get_docker_networks(networks, network_ids):
return parsed_networks or []
def get_nanoseconds_from_raw_option(name, value):
def get_nanoseconds_from_raw_option(name: str, value: t.Any) -> int | None:
if value is None:
return None
if isinstance(value, int):
@ -1003,12 +1009,14 @@ def get_nanoseconds_from_raw_option(name, value):
)
def get_value(key, values, default=None):
def get_value(key: str, values: dict[str, t.Any], default: t.Any = None) -> t.Any:
value = values.get(key)
return value if value is not None else default
def has_dict_changed(new_dict, old_dict):
def has_dict_changed(
new_dict: dict[str, t.Any] | None, old_dict: dict[str, t.Any] | None
) -> bool:
"""
Check if new_dict has differences compared to old_dict while
ignoring keys in old_dict which are None in new_dict.
@ -1019,6 +1027,9 @@ def has_dict_changed(new_dict, old_dict):
return True
if not old_dict and new_dict:
return True
if old_dict is None:
# in this case new_dict is empty, only the type checker didn't notice
return False
defined_options = {
option: value for option, value in new_dict.items() if value is not None
}
@ -1031,12 +1042,17 @@ def has_dict_changed(new_dict, old_dict):
return False
def has_list_changed(new_list, old_list, sort_lists=True, sort_key=None):
def has_list_changed(
new_list: list[t.Any] | None,
old_list: list[t.Any] | None,
sort_lists: bool = True,
sort_key: str | None = None,
) -> bool:
"""
Check two lists have differences. Sort lists by default.
"""
def sort_list(unsorted_list):
def sort_list(unsorted_list: list[t.Any]) -> list[t.Any]:
"""
Sort a given list.
The list may contain dictionaries, so use the sort key to handle them.
@ -1093,7 +1109,10 @@ def has_list_changed(new_list, old_list, sort_lists=True, sort_key=None):
return False
def have_networks_changed(new_networks, old_networks):
def have_networks_changed(
new_networks: list[dict[str, t.Any]] | None,
old_networks: list[dict[str, t.Any]] | None,
) -> bool:
"""Special case list checking for networks to sort aliases"""
if new_networks is None:
@ -1123,68 +1142,72 @@ def have_networks_changed(new_networks, old_networks):
class DockerService(DockerBaseClass):
def __init__(self, docker_api_version, docker_py_version):
def __init__(
self, docker_api_version: LooseVersion, docker_py_version: LooseVersion
) -> None:
super().__init__()
self.image = ""
self.command = None
self.args = None
self.endpoint_mode = None
self.dns = None
self.healthcheck = None
self.healthcheck_disabled = None
self.hostname = None
self.hosts = None
self.tty = None
self.dns_search = None
self.dns_options = None
self.env = None
self.force_update = None
self.groups = None
self.log_driver = None
self.log_driver_options = None
self.labels = None
self.container_labels = None
self.sysctls = None
self.limit_cpu = None
self.limit_memory = None
self.reserve_cpu = None
self.reserve_memory = None
self.mode = "replicated"
self.user = None
self.mounts = None
self.configs = None
self.secrets = None
self.constraints = None
self.replicas_max_per_node = None
self.networks = None
self.stop_grace_period = None
self.stop_signal = None
self.publish = None
self.placement_preferences = None
self.replicas = -1
self.image: str | None = ""
self.command: t.Any = None
self.args: list[str] | None = None
self.endpoint_mode: t.Literal["vip", "dnsrr"] | None = None
self.dns: list[str] | None = None
self.healthcheck: dict[str, t.Any] | None = None
self.healthcheck_disabled: bool | None = None
self.hostname: str | None = None
self.hosts: dict[str, t.Any] | None = None
self.tty: bool | None = None
self.dns_search: list[str] | None = None
self.dns_options: list[str] | None = None
self.env: t.Any = None
self.force_update: int | None = None
self.groups: list[str] | None = None
self.log_driver: str | None = None
self.log_driver_options: dict[str, t.Any] | None = None
self.labels: dict[str, t.Any] | None = None
self.container_labels: dict[str, t.Any] | None = None
self.sysctls: dict[str, t.Any] | None = None
self.limit_cpu: float | None = None
self.limit_memory: int | None = None
self.reserve_cpu: float | None = None
self.reserve_memory: int | None = None
self.mode: t.Literal["replicated", "global", "replicated-job"] = "replicated"
self.user: str | None = None
self.mounts: list[dict[str, t.Any]] | None = None
self.configs: list[dict[str, t.Any]] | None = None
self.secrets: list[dict[str, t.Any]] | None = None
self.constraints: list[str] | None = None
self.replicas_max_per_node: int | None = None
self.networks: list[t.Any] | None = None
self.stop_grace_period: int | None = None
self.stop_signal: str | None = None
self.publish: list[dict[str, t.Any]] | None = None
self.placement_preferences: list[dict[str, t.Any]] | None = None
self.replicas: int | None = -1
self.service_id = False
self.service_version = False
self.read_only = None
self.restart_policy = None
self.restart_policy_attempts = None
self.restart_policy_delay = None
self.restart_policy_window = None
self.rollback_config = None
self.update_delay = None
self.update_parallelism = None
self.update_failure_action = None
self.update_monitor = None
self.update_max_failure_ratio = None
self.update_order = None
self.working_dir = None
self.init = None
self.cap_add = None
self.cap_drop = None
self.read_only: bool | None = None
self.restart_policy: t.Literal["none", "on-failure", "any"] | None = None
self.restart_policy_attempts: int | None = None
self.restart_policy_delay: str | None = None
self.restart_policy_window: str | None = None
self.rollback_config: dict[str, t.Any] | None = None
self.update_delay: str | None = None
self.update_parallelism: int | None = None
self.update_failure_action: (
t.Literal["continue", "pause", "rollback"] | None
) = None
self.update_monitor: str | None = None
self.update_max_failure_ratio: float | None = None
self.update_order: str | None = None
self.working_dir: str | None = None
self.init: bool | None = None
self.cap_add: list[str] | None = None
self.cap_drop: list[str] | None = None
self.docker_api_version = docker_api_version
self.docker_py_version = docker_py_version
def get_facts(self):
def get_facts(self) -> dict[str, t.Any]:
return {
"image": self.image,
"mounts": self.mounts,
@ -1242,19 +1265,21 @@ class DockerService(DockerBaseClass):
}
@property
def can_update_networks(self):
def can_update_networks(self) -> bool:
# Before Docker API 1.29 adding/removing networks was not supported
return self.docker_api_version >= LooseVersion(
"1.29"
) and self.docker_py_version >= LooseVersion("2.7")
@property
def can_use_task_template_networks(self):
def can_use_task_template_networks(self) -> bool:
# In Docker API 1.25 attaching networks to TaskTemplate is preferred over Spec
return self.docker_py_version >= LooseVersion("2.7")
@staticmethod
def get_restart_config_from_ansible_params(params):
def get_restart_config_from_ansible_params(
params: dict[str, t.Any],
) -> dict[str, t.Any]:
restart_config = params["restart_config"] or {}
condition = get_value(
"condition",
@ -1282,7 +1307,9 @@ class DockerService(DockerBaseClass):
}
@staticmethod
def get_update_config_from_ansible_params(params):
def get_update_config_from_ansible_params(
params: dict[str, t.Any],
) -> dict[str, t.Any]:
update_config = params["update_config"] or {}
parallelism = get_value(
"parallelism",
@ -1320,7 +1347,9 @@ class DockerService(DockerBaseClass):
}
@staticmethod
def get_rollback_config_from_ansible_params(params):
def get_rollback_config_from_ansible_params(
params: dict[str, t.Any],
) -> dict[str, t.Any] | None:
if params["rollback_config"] is None:
return None
rollback_config = params["rollback_config"] or {}
@ -1340,7 +1369,7 @@ class DockerService(DockerBaseClass):
}
@staticmethod
def get_logging_from_ansible_params(params):
def get_logging_from_ansible_params(params: dict[str, t.Any]) -> dict[str, t.Any]:
logging_config = params["logging"] or {}
driver = get_value(
"driver",
@ -1356,7 +1385,7 @@ class DockerService(DockerBaseClass):
}
@staticmethod
def get_limits_from_ansible_params(params):
def get_limits_from_ansible_params(params: dict[str, t.Any]) -> dict[str, t.Any]:
limits = params["limits"] or {}
cpus = get_value(
"cpus",
@ -1379,7 +1408,9 @@ class DockerService(DockerBaseClass):
}
@staticmethod
def get_reservations_from_ansible_params(params):
def get_reservations_from_ansible_params(
params: dict[str, t.Any],
) -> dict[str, t.Any]:
reservations = params["reservations"] or {}
cpus = get_value(
"cpus",
@ -1403,7 +1434,7 @@ class DockerService(DockerBaseClass):
}
@staticmethod
def get_placement_from_ansible_params(params):
def get_placement_from_ansible_params(params: dict[str, t.Any]) -> dict[str, t.Any]:
placement = params["placement"] or {}
constraints = get_value("constraints", placement)
@ -1419,14 +1450,14 @@ class DockerService(DockerBaseClass):
@classmethod
def from_ansible_params(
cls,
ap,
ap: dict[str, t.Any],
old_service,
image_digest,
secret_ids,
config_ids,
network_ids,
client,
):
secret_ids: dict[str, str],
config_ids: dict[str, str],
network_ids: dict[str, str],
client: AnsibleDockerClient,
) -> DockerService:
s = DockerService(client.docker_api_version, client.docker_py_version)
s.image = image_digest
s.args = ap["args"]
@ -1596,7 +1627,7 @@ class DockerService(DockerBaseClass):
return s
def compare(self, os):
def compare(self, os: DockerService) -> tuple[bool, DifferenceTracker, bool, bool]:
differences = DifferenceTracker()
needs_rebuild = False
force_update = False
@ -1784,7 +1815,7 @@ class DockerService(DockerBaseClass):
differences.add(
"update_order", parameter=self.update_order, active=os.update_order
)
has_image_changed, change = self.has_image_changed(os.image)
has_image_changed, change = self.has_image_changed(os.image or "")
if has_image_changed:
differences.add("image", parameter=self.image, active=change)
if self.user and self.user != os.user:
@ -1828,7 +1859,7 @@ class DockerService(DockerBaseClass):
force_update,
)
def has_healthcheck_changed(self, old_publish):
def has_healthcheck_changed(self, old_publish: DockerService) -> bool:
if self.healthcheck_disabled is False and self.healthcheck is None:
return False
if self.healthcheck_disabled:
@ -1838,14 +1869,14 @@ class DockerService(DockerBaseClass):
return False
return self.healthcheck != old_publish.healthcheck
def has_publish_changed(self, old_publish):
def has_publish_changed(self, old_publish: list[dict[str, t.Any]] | None) -> bool:
if self.publish is None:
return False
old_publish = old_publish or []
if len(self.publish) != len(old_publish):
return True
def publish_sorter(item):
def publish_sorter(item: dict[str, t.Any]) -> tuple[int, int, str]:
return (
item.get("published_port") or 0,
item.get("target_port") or 0,
@ -1869,12 +1900,13 @@ class DockerService(DockerBaseClass):
return True
return False
def has_image_changed(self, old_image):
def has_image_changed(self, old_image: str) -> tuple[bool, str]:
assert self.image is not None
if "@" not in self.image:
old_image = old_image.split("@")[0]
return self.image != old_image, old_image
def build_container_spec(self):
def build_container_spec(self) -> types.ContainerSpec:
mounts = None
if self.mounts is not None:
mounts = []
@ -1945,7 +1977,7 @@ class DockerService(DockerBaseClass):
secrets.append(types.SecretReference(**secret_args))
dns_config_args = {}
dns_config_args: dict[str, t.Any] = {}
if self.dns is not None:
dns_config_args["nameservers"] = self.dns
if self.dns_search is not None:
@ -1954,7 +1986,7 @@ class DockerService(DockerBaseClass):
dns_config_args["options"] = self.dns_options
dns_config = types.DNSConfig(**dns_config_args) if dns_config_args else None
container_spec_args = {}
container_spec_args: dict[str, t.Any] = {}
if self.command is not None:
container_spec_args["command"] = self.command
if self.args is not None:
@ -2004,8 +2036,8 @@ class DockerService(DockerBaseClass):
return types.ContainerSpec(self.image, **container_spec_args)
def build_placement(self):
placement_args = {}
def build_placement(self) -> types.Placement | None:
placement_args: dict[str, t.Any] = {}
if self.constraints is not None:
placement_args["constraints"] = self.constraints
if self.replicas_max_per_node is not None:
@ -2018,8 +2050,8 @@ class DockerService(DockerBaseClass):
]
return types.Placement(**placement_args) if placement_args else None
def build_update_config(self):
update_config_args = {}
def build_update_config(self) -> types.UpdateConfig | None:
update_config_args: dict[str, t.Any] = {}
if self.update_parallelism is not None:
update_config_args["parallelism"] = self.update_parallelism
if self.update_delay is not None:
@ -2034,16 +2066,16 @@ class DockerService(DockerBaseClass):
update_config_args["order"] = self.update_order
return types.UpdateConfig(**update_config_args) if update_config_args else None
def build_log_driver(self):
log_driver_args = {}
def build_log_driver(self) -> types.DriverConfig | None:
log_driver_args: dict[str, t.Any] = {}
if self.log_driver is not None:
log_driver_args["name"] = self.log_driver
if self.log_driver_options is not None:
log_driver_args["options"] = self.log_driver_options
return types.DriverConfig(**log_driver_args) if log_driver_args else None
def build_restart_policy(self):
restart_policy_args = {}
def build_restart_policy(self) -> types.RestartPolicy | None:
restart_policy_args: dict[str, t.Any] = {}
if self.restart_policy is not None:
restart_policy_args["condition"] = self.restart_policy
if self.restart_policy_delay is not None:
@ -2056,7 +2088,7 @@ class DockerService(DockerBaseClass):
types.RestartPolicy(**restart_policy_args) if restart_policy_args else None
)
def build_rollback_config(self):
def build_rollback_config(self) -> types.RollbackConfig | None:
if self.rollback_config is None:
return None
rollback_config_options = [
@ -2078,8 +2110,8 @@ class DockerService(DockerBaseClass):
else None
)
def build_resources(self):
resources_args = {}
def build_resources(self) -> types.Resources | None:
resources_args: dict[str, t.Any] = {}
if self.limit_cpu is not None:
resources_args["cpu_limit"] = int(self.limit_cpu * 1000000000.0)
if self.limit_memory is not None:
@ -2090,12 +2122,16 @@ class DockerService(DockerBaseClass):
resources_args["mem_reservation"] = self.reserve_memory
return types.Resources(**resources_args) if resources_args else None
def build_task_template(self, container_spec, placement=None):
def build_task_template(
self,
container_spec: types.ContainerSpec,
placement: types.Placement | None = None,
) -> types.TaskTemplate:
log_driver = self.build_log_driver()
restart_policy = self.build_restart_policy()
resources = self.build_resources()
task_template_args = {}
task_template_args: dict[str, t.Any] = {}
if placement is not None:
task_template_args["placement"] = placement
if log_driver is not None:
@ -2112,12 +2148,12 @@ class DockerService(DockerBaseClass):
task_template_args["networks"] = networks
return types.TaskTemplate(container_spec=container_spec, **task_template_args)
def build_service_mode(self):
def build_service_mode(self) -> types.ServiceMode:
if self.mode == "global":
self.replicas = None
return types.ServiceMode(self.mode, replicas=self.replicas)
def build_networks(self):
def build_networks(self) -> list[dict[str, t.Any]] | None:
networks = None
if self.networks is not None:
networks = []
@ -2130,8 +2166,8 @@ class DockerService(DockerBaseClass):
networks.append(docker_network)
return networks
def build_endpoint_spec(self):
endpoint_spec_args = {}
def build_endpoint_spec(self) -> types.EndpointSpec | None:
endpoint_spec_args: dict[str, t.Any] = {}
if self.publish is not None:
ports = []
for port in self.publish:
@ -2149,7 +2185,7 @@ class DockerService(DockerBaseClass):
endpoint_spec_args["mode"] = self.endpoint_mode
return types.EndpointSpec(**endpoint_spec_args) if endpoint_spec_args else None
def build_docker_service(self):
def build_docker_service(self) -> dict[str, t.Any]:
container_spec = self.build_container_spec()
placement = self.build_placement()
task_template = self.build_task_template(container_spec, placement)
@ -2159,7 +2195,10 @@ class DockerService(DockerBaseClass):
service_mode = self.build_service_mode()
endpoint_spec = self.build_endpoint_spec()
service = {"task_template": task_template, "mode": service_mode}
service: dict[str, t.Any] = {
"task_template": task_template,
"mode": service_mode,
}
if update_config:
service["update_config"] = update_config
if rollback_config:
@ -2176,13 +2215,12 @@ class DockerService(DockerBaseClass):
class DockerServiceManager:
def __init__(self, client):
def __init__(self, client: AnsibleDockerClient):
self.client = client
self.retries = 2
self.diff_tracker = None
self.diff_tracker: DifferenceTracker | None = None
def get_service(self, name):
def get_service(self, name: str) -> DockerService | None:
try:
raw_data = self.client.inspect_service(name)
except NotFound:
@ -2415,7 +2453,9 @@ class DockerServiceManager:
ds.init = task_template_data["ContainerSpec"].get("Init", False)
return ds
def update_service(self, name, old_service, new_service):
def update_service(
self, name: str, old_service: DockerService, new_service: DockerService
) -> None:
service_data = new_service.build_docker_service()
result = self.client.update_service(
old_service.service_id,
@ -2427,15 +2467,15 @@ class DockerServiceManager:
# (see https://github.com/docker/docker-py/pull/2272)
self.client.report_warnings(result, ["Warning"])
def create_service(self, name, service):
def create_service(self, name: str, service: DockerService) -> None:
service_data = service.build_docker_service()
result = self.client.create_service(name=name, **service_data)
self.client.report_warnings(result, ["Warning"])
def remove_service(self, name):
def remove_service(self, name: str) -> None:
self.client.remove_service(name)
def get_image_digest(self, name, resolve=False):
def get_image_digest(self, name: str, resolve: bool = False) -> str:
if not name or not resolve:
return name
repo, tag = parse_repository_tag(name)
@ -2446,10 +2486,10 @@ class DockerServiceManager:
digest = distribution_data["Descriptor"]["digest"]
return f"{name}@{digest}"
def get_networks_names_ids(self):
def get_networks_names_ids(self) -> dict[str, str]:
return {network["Name"]: network["Id"] for network in self.client.networks()}
def get_missing_secret_ids(self):
def get_missing_secret_ids(self) -> dict[str, str]:
"""
Resolve missing secret ids by looking them up by name
"""
@ -2471,7 +2511,7 @@ class DockerServiceManager:
self.client.fail(f'Could not find a secret named "{secret_name}"')
return secrets
def get_missing_config_ids(self):
def get_missing_config_ids(self) -> dict[str, str]:
"""
Resolve missing config ids by looking them up by name
"""
@ -2493,7 +2533,7 @@ class DockerServiceManager:
self.client.fail(f'Could not find a config named "{config_name}"')
return configs
def run(self):
def run(self) -> tuple[str, bool, bool, list[str], dict[str, t.Any]]:
self.diff_tracker = DifferenceTracker()
module = self.client.module
@ -2582,7 +2622,7 @@ class DockerServiceManager:
return msg, changed, rebuilt, differences.get_legacy_docker_diffs(), facts
def run_safe(self):
def run_safe(self) -> tuple[str, bool, bool, list[str], dict[str, t.Any]]:
while True:
try:
return self.run()
@ -2596,20 +2636,20 @@ class DockerServiceManager:
raise
def _detect_publish_mode_usage(client):
def _detect_publish_mode_usage(client: AnsibleDockerClient) -> bool:
for publish_def in client.module.params["publish"] or []:
if publish_def.get("mode"):
return True
return False
def _detect_healthcheck_start_period(client):
def _detect_healthcheck_start_period(client: AnsibleDockerClient) -> bool:
if client.module.params["healthcheck"]:
return client.module.params["healthcheck"]["start_period"] is not None
return False
def _detect_mount_tmpfs_usage(client):
def _detect_mount_tmpfs_usage(client: AnsibleDockerClient) -> bool:
for mount in client.module.params["mounts"] or []:
if mount.get("type") == "tmpfs":
return True
@ -2620,14 +2660,14 @@ def _detect_mount_tmpfs_usage(client):
return False
def _detect_update_config_failure_action_rollback(client):
def _detect_update_config_failure_action_rollback(client: AnsibleDockerClient) -> bool:
rollback_config_failure_action = (client.module.params["update_config"] or {}).get(
"failure_action"
)
return rollback_config_failure_action == "rollback"
def main():
def main() -> None:
argument_spec = {
"name": {"type": "str", "required": True},
"image": {"type": "str"},
@ -2948,6 +2988,7 @@ def main():
"swarm_service": facts,
}
if client.module._diff:
assert dsm.diff_tracker is not None
before, after = dsm.diff_tracker.get_before_after()
results["diff"] = {"before": before, "after": after}

5
plugins/modules/docker_swarm_service_info.py
View File

@ -63,6 +63,7 @@ service:
"""
import traceback
import typing as t
try:
@ -79,12 +80,12 @@ from ansible_collections.community.docker.plugins.module_utils._swarm import (
)
def get_service_info(client):
def get_service_info(client: AnsibleDockerSwarmClient) -> dict[str, t.Any] | None:
service = client.module.params["name"]
return client.get_service_inspect(service_id=service, skip_missing=True)
def main():
def main() -> None:
argument_spec = {
"name": {"type": "str", "required": True},
}

1
tests/sanity/ignore-2.17.txt
View File

@ -19,4 +19,5 @@ plugins/modules/docker_image.py no-assert
plugins/modules/docker_image_tag.py no-assert
plugins/modules/docker_login.py no-assert
plugins/modules/docker_plugin.py no-assert
plugins/modules/docker_swarm_service.py no-assert
plugins/modules/docker_volume.py no-assert

1
tests/sanity/ignore-2.18.txt
View File

@ -18,4 +18,5 @@ plugins/modules/docker_image.py no-assert
plugins/modules/docker_image_tag.py no-assert
plugins/modules/docker_login.py no-assert
plugins/modules/docker_plugin.py no-assert
plugins/modules/docker_swarm_service.py no-assert
plugins/modules/docker_volume.py no-assert

1
tests/sanity/ignore-2.19.txt
View File

@ -12,4 +12,5 @@ plugins/modules/docker_image.py no-assert
plugins/modules/docker_image_tag.py no-assert
plugins/modules/docker_login.py no-assert
plugins/modules/docker_plugin.py no-assert
plugins/modules/docker_swarm_service.py no-assert
plugins/modules/docker_volume.py no-assert

1
tests/sanity/ignore-2.20.txt
View File

@ -12,4 +12,5 @@ plugins/modules/docker_image.py no-assert
plugins/modules/docker_image_tag.py no-assert
plugins/modules/docker_login.py no-assert
plugins/modules/docker_plugin.py no-assert
plugins/modules/docker_swarm_service.py no-assert
plugins/modules/docker_volume.py no-assert

1
tests/sanity/ignore-2.21.txt
View File

@ -12,4 +12,5 @@ plugins/modules/docker_image.py no-assert
plugins/modules/docker_image_tag.py no-assert
plugins/modules/docker_login.py no-assert
plugins/modules/docker_plugin.py no-assert
plugins/modules/docker_swarm_service.py no-assert
plugins/modules/docker_volume.py no-assert