mirror of
https://github.com/ansible-collections/community.general.git
synced 2026-04-08 21:17:20 +00:00
Add more types as suggested.
This commit is contained in:
parent
dc43e5992f
commit
c3bd136c6b
2 changed files with 37 additions and 18 deletions
|
|
@ -50,11 +50,11 @@ class BtrfsCommands:
|
|||
current["devices"].append(self.__parse_filesystem_device(line))
|
||||
return filesystems
|
||||
|
||||
def __parse_filesystem(self, line):
|
||||
def __parse_filesystem(self, line) -> dict[str, t.Any]:
|
||||
label = re.sub(r"\s*uuid:.*$", "", re.sub(r"^Label:\s*", "", line))
|
||||
id = re.sub(r"^.*uuid:\s*", "", line)
|
||||
|
||||
filesystem = {}
|
||||
filesystem: dict[str, t.Any] = {}
|
||||
filesystem["label"] = label.strip("'") if label != "none" else None
|
||||
filesystem["uuid"] = id
|
||||
filesystem["devices"] = []
|
||||
|
|
@ -63,10 +63,10 @@ class BtrfsCommands:
|
|||
filesystem["default_subvolid"] = None
|
||||
return filesystem
|
||||
|
||||
def __parse_filesystem_device(self, line):
|
||||
def __parse_filesystem_device(self, line: str) -> str:
|
||||
return re.sub(r"^.*path\s", "", line)
|
||||
|
||||
def subvolumes_list(self, filesystem_path):
|
||||
def subvolumes_list(self, filesystem_path: str) -> list[dict[str, t.Any]]:
|
||||
command = f"{self.__btrfs} subvolume list -tap {filesystem_path}"
|
||||
result = self.__module.run_command(command, check_rc=True)
|
||||
stdout = [x.split("\t") for x in result[1].splitlines()]
|
||||
|
|
|
|||
|
|
@ -5,12 +5,16 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import typing as t
|
||||
|
||||
from ansible.module_utils.basic import AnsibleModule
|
||||
from ansible.module_utils.common.collections import is_sequence
|
||||
from ansible.module_utils.common.locale import get_best_parsable_locale
|
||||
from ansible_collections.community.general.plugins.module_utils import cmd_runner_fmt
|
||||
|
||||
if t.TYPE_CHECKING:
|
||||
from collections.abc import Callable, Sequence
|
||||
|
||||
|
||||
def _ensure_list(value):
|
||||
return list(value) if is_sequence(value) else [value]
|
||||
|
|
@ -25,7 +29,7 @@ class CmdRunnerException(Exception):
|
|||
|
||||
|
||||
class MissingArgumentFormat(CmdRunnerException):
|
||||
def __init__(self, arg, args_order, args_formats):
|
||||
def __init__(self, arg, args_order: tuple[str, ...], args_formats) -> None:
|
||||
self.args_order = args_order
|
||||
self.arg = arg
|
||||
self.args_formats = args_formats
|
||||
|
|
@ -38,7 +42,7 @@ class MissingArgumentFormat(CmdRunnerException):
|
|||
|
||||
|
||||
class MissingArgumentValue(CmdRunnerException):
|
||||
def __init__(self, args_order, arg):
|
||||
def __init__(self, args_order: tuple[str, ...], arg) -> None:
|
||||
self.args_order = args_order
|
||||
self.arg = arg
|
||||
|
||||
|
|
@ -73,19 +77,19 @@ class CmdRunner:
|
|||
"""
|
||||
|
||||
@staticmethod
|
||||
def _prepare_args_order(order):
|
||||
return tuple(order) if is_sequence(order) else tuple(order.split())
|
||||
def _prepare_args_order(order: str | Sequence[str]) -> tuple[str, ...]:
|
||||
return tuple(order) if is_sequence(order) else tuple(order.split()) # type: ignore
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
module: AnsibleModule,
|
||||
command,
|
||||
arg_formats=None,
|
||||
default_args_order=(),
|
||||
check_rc=False,
|
||||
force_lang="C",
|
||||
path_prefix=None,
|
||||
environ_update=None,
|
||||
arg_formats: dict[str, Callable] | None = None,
|
||||
default_args_order: str | Sequence[str] = (),
|
||||
check_rc: bool = False,
|
||||
force_lang: str = "C",
|
||||
path_prefix: list[str] | None = None,
|
||||
environ_update: dict[str, str] | None = None,
|
||||
):
|
||||
self.module = module
|
||||
self.command = _ensure_list(command)
|
||||
|
|
@ -118,10 +122,17 @@ class CmdRunner:
|
|||
)
|
||||
|
||||
@property
|
||||
def binary(self):
|
||||
def binary(self) -> str:
|
||||
return self.command[0]
|
||||
|
||||
def __call__(self, args_order=None, output_process=None, check_mode_skip=False, check_mode_return=None, **kwargs):
|
||||
def __call__(
|
||||
self,
|
||||
args_order: str | Sequence[str] | None = None,
|
||||
output_process=None,
|
||||
check_mode_skip: bool = False,
|
||||
check_mode_return=None,
|
||||
**kwargs,
|
||||
):
|
||||
if output_process is None:
|
||||
output_process = _process_as_is
|
||||
if args_order is None:
|
||||
|
|
@ -147,7 +158,15 @@ class CmdRunner:
|
|||
|
||||
|
||||
class _CmdRunnerContext:
|
||||
def __init__(self, runner, args_order, output_process, check_mode_skip, check_mode_return, **kwargs):
|
||||
def __init__(
|
||||
self,
|
||||
runner: CmdRunner,
|
||||
args_order: tuple[str, ...],
|
||||
output_process,
|
||||
check_mode_skip: bool,
|
||||
check_mode_return,
|
||||
**kwargs,
|
||||
) -> None:
|
||||
self.runner = runner
|
||||
self.args_order = tuple(args_order)
|
||||
self.output_process = output_process
|
||||
|
|
@ -205,7 +224,7 @@ class _CmdRunnerContext:
|
|||
return self.results_processed
|
||||
|
||||
@property
|
||||
def run_info(self):
|
||||
def run_info(self) -> dict[str, t.Any]:
|
||||
return dict(
|
||||
check_rc=self.check_rc,
|
||||
environ_update=self.environ_update,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue