Resolve mypy disallow_any_generics

Related: ansible/ansible-runner#1340

pyproject.toml

@@ -72,9 +72,16 @@ color_output = true
 error_summary = true
 # disallow_untyped_calls = true
 # disallow_untyped_defs = true
-# disallow_any_generics = true
+disallow_any_generics = true
 incremental = false
 
+[[tool.mypy.overrides]]
+# https://github.com/ansible/ansible-runner/issues/1340
+module = [
+  "ansible_runner"
+]
+ignore_missing_imports = true
+
 [tool.pydoclint]
 allow-init-docstring = true
 arg-type-hints-in-docstring = false

src/ansible_navigator/command_runner/command_runner.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass
 from dataclasses import field
 from queue import Queue
+from typing import Any
 
 from ansible_navigator.utils.definitions import LogMessage
 
@@ -27,11 +28,11 @@ class Command:
     # pylint: disable=too-many-instance-attributes
     identity: str
     command: str
-    post_process: Callable
+    post_process: Callable[..., Any]
     return_code: int = 0
     stdout: str = ""
     stderr: str = ""
-    details: list = field(default_factory=list)
+    details: list[Any] = field(default_factory=list)
     errors: str = ""
     messages: list[LogMessage] = field(default_factory=list)
 
@@ -73,7 +74,9 @@ def run_command(command: Command) -> None:
         command.stderr = str(exc.stderr)
 
 
-def worker(pending_queue: multiprocessing.Queue, completed_queue: multiprocessing.Queue) -> None:
+def worker(
+    pending_queue: multiprocessing.Queue[Any], completed_queue: multiprocessing.Queue[Any]
+) -> None:
     """Read pending, run, post process, and place in completed.
 
     :param pending_queue: All pending commands

src/ansible_navigator/configuration_subsystem/definitions.py

@@ -467,10 +467,12 @@ class PaeChangeRequest:
 # and some common ones
 
 # A type used for the settings as a dictionary
-SettingsFileType = NewType("SettingsFileType", dict[str, bool | dict | int | str | list])
+SettingsFileType = NewType(
+    "SettingsFileType", dict[str, bool | dict[Any, Any] | int | str | list[Any]]
+)
 
 # A type used to describe a schema file for the settings
 SettingsSchemaType = NewType(
     "SettingsSchemaType",
-    dict[str, dict[str, bool | dict | int | str | list]],
+    dict[str, dict[str, bool | dict[Any, Any] | int | str | list[Any]]],
 )

src/ansible_navigator/configuration_subsystem/navigator_post_processor.py

@@ -74,7 +74,7 @@ class NavigatorPostProcessor:
     # pylint:disable=too-many-public-methods
     """Application post processor."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Initialize the post processor."""
         #: Volume mounts accumulated from post processing various config entries.
         #: These get processed towards the end, in the (delayed)

src/ansible_navigator/configuration_subsystem/utils.py

@@ -22,7 +22,7 @@
 
 def create_settings_file_sample(
     settings_path: str,
-    placeholder: bool | int | str | dict | list = "",
+    placeholder: bool | int | str | dict[Any, Any] | list[Any] = "",
 ) -> SettingsFileType:
     """Generate a settings file sample.
 

src/ansible_navigator/content_defs.py

@@ -57,7 +57,7 @@ def asdict(
         self,
         content_view: ContentView,
         serialization_format: SerializationFormat,
-    ) -> DictType:
+    ) -> DictType[Any]:
         """Convert thy self into a dictionary.
 
         :param content_view: The content view
@@ -77,28 +77,28 @@ def asdict(
             return asdict(self)
         return dump_self_as_dict()
 
-    def serialize_json_full(self) -> DictType:
+    def serialize_json_full(self) -> DictType[Any]:
         """Provide dictionary for ``JSON`` with all attributes.
 
         :returns: A dictionary created from self
         """
         return asdict(self)
 
-    def serialize_json_normal(self) -> DictType:
+    def serialize_json_normal(self) -> DictType[Any]:
         """Provide dictionary for ``JSON`` with curated attributes.
 
         :returns: A dictionary created from self
         """
         return asdict(self)
 
-    def serialize_yaml_full(self) -> DictType:
+    def serialize_yaml_full(self) -> DictType[Any]:
         """Provide dictionary for ``YAML`` with all attributes.
 
         :returns: A dictionary created from self
         """
         return asdict(self)
 
-    def serialize_yaml_normal(self) -> DictType:
+    def serialize_yaml_normal(self) -> DictType[Any]:
         """Provide dictionary for ``JSON`` with curated attributes.
 
         :returns: A dictionary created from self
@@ -131,8 +131,8 @@ def items(self):
         return asdict(self).items()
 
 
-ContentTypeSingle = bool | float | int | str | dict[str, Any] | ContentBase
-ContentTypeSequence = list[Any] | Sequence[ContentBase]
+ContentTypeSingle = bool | float | int | str | dict[str, Any] | ContentBase[Any]
+ContentTypeSequence = list[Any] | Sequence[ContentBase[Any]]
 ContentType = ContentTypeSingle | ContentTypeSequence
 
 

src/ansible_navigator/data/catalog_collections.py

@@ -58,11 +58,11 @@ def __init__(self, directories: list[Path]):
         :param directories: A list of directories that may contain collections
         """
         self._directories: list[Path] = directories
-        self._collections: OrderedDict[str, dict] = OrderedDict()
+        self._collections: OrderedDict[str, dict[Any, Any]] = OrderedDict()
         self._errors: list[dict[str, str]] = []
         self._messages: list[str] = []
 
-    def _catalog_plugins(self, collection: dict) -> None:
+    def _catalog_plugins(self, collection: dict[Any, Any]) -> None:
         """Catalog the plugins within a collection.
 
         :param collection: Details describing the collection
@@ -202,7 +202,7 @@ def _catalog_roles(self, collection: dict[str, Any]) -> None:
                 collection["roles"].append(role)
 
     @staticmethod
-    def _generate_checksum(file_path: Path, relative_path: Path) -> dict:
+    def _generate_checksum(file_path: Path, relative_path: Path) -> dict[str, Any]:
         """Generate a standard checksum for a file.
 
         :param file_path: The path to the file to generate a checksum for
@@ -226,8 +226,8 @@ def _process_plugin_dir(
         self,
         plugin_type: str,
         filenames: Generator[Path, None, None],
-        file_checksums: dict[str, dict],
-        collection: dict,
+        file_checksums: dict[str, dict[Any, Any]],
+        collection: dict[Any, Any],
     ) -> None:
         """Process each plugin within one plugin directory.
 
@@ -321,7 +321,7 @@ def _find_shadows(self) -> None:
                             i_collection["path"],
                         )
 
-    def process_directories(self) -> tuple[dict, list]:
+    def process_directories(self) -> tuple[dict[Any, Any], list[dict[str, str]]]:
         """Process each parent directory.
 
         :returns: All collections found and any errors
@@ -339,7 +339,7 @@ def process_directories(self) -> tuple[dict, list]:
 
     def add_pseudo_builtin(self) -> None:
         """Add the pseudo builtin collection."""
-        collection: dict[str, str | list | dict] = {}
+        collection: dict[str, str | list[str] | dict[Any, Any]] = {}
         collection["known_as"] = "ansible.builtin"
         collection["plugin_checksums"] = {}
         collection["path"] = str(Path(plugins.__file__).parents[1])
@@ -351,7 +351,9 @@ def add_pseudo_builtin(self) -> None:
         self._messages.append(msg)
 
 
-def worker(pending_queue: multiprocessing.Queue, completed_queue: multiprocessing.Queue) -> None:
+def worker(
+    pending_queue: multiprocessing.Queue[Any], completed_queue: multiprocessing.Queue[Any]
+) -> None:
     """Extract the documentation from a plugin, place in completed queue.
 
     :param pending_queue: A queue with plugins to process
@@ -402,7 +404,9 @@ def worker(pending_queue: multiprocessing.Queue, completed_queue: multiprocessin
             completed_queue.put(("error", (checksum, plugin_path, err_message)))
 
 
-def identify_missing(collections: dict, collection_cache: KeyValueStore) -> tuple[set, list, int]:
+def identify_missing(
+    collections: dict[Any, Any], collection_cache: KeyValueStore
+) -> tuple[set[Any], list[Any], int]:
     """Identify plugins missing from the cache.
 
     :param collections: All plugins found across all collections
@@ -466,7 +470,7 @@ def parse_args() -> tuple[argparse.Namespace, list[Path]]:
     return parsed_args, resolved
 
 
-def retrieve_collections_paths() -> dict:
+def retrieve_collections_paths() -> dict[Any, Any]:
     """Retrieve the currently set collection paths.
 
     :returns: Errors or the configured collection directories
@@ -488,9 +492,9 @@ def retrieve_collections_paths() -> dict:
 
 def retrieve_docs(
     collection_cache: KeyValueStore,
-    errors: list,
-    missing: list,
-    stats: dict,
+    errors: list[dict[str, str]],
+    missing: list[str],
+    stats: dict[Any, Any],
 ) -> None:
     # pylint: disable=too-many-locals
     """Extract the docs from the plugins.
@@ -528,7 +532,7 @@ def retrieve_docs(
             stats["cache_added_errors"] += 1
 
 
-def run_command(cmd: list) -> dict:
+def run_command(cmd: list[str]) -> dict[str, str]:
     """Run a command using subprocess.
 
     :param cmd: The command to run, split
@@ -547,7 +551,7 @@ def run_command(cmd: list) -> dict:
         return {"error": str(exc)}
 
 
-def main() -> dict:
+def main() -> dict[Any, Any]:
     # pylint: disable=protected-access
     # pylint: disable=used-before-assignment
     """Run the collection catalog process.

src/ansible_navigator/data/image_introspect.py

@@ -13,17 +13,19 @@
 from queue import Queue
 from types import SimpleNamespace
 from typing import Any
+from typing import TypeAlias
 
 
-JSONTypes = bool | int | str | dict | list
+# https://github.com/python/typing/issues/182#issuecomment-1320974824
+JSONTypes: TypeAlias = dict[str, "JSONTypes"] | list["JSONTypes"] | str | int | float | bool | None
 
 
 class Command(SimpleNamespace):
     """Abstraction for a details about a shell command."""
 
     id_: str
     command: str
-    parse: Callable
+    parse: Callable[..., Any]
     stdout: str = ""
     stderr: str = ""
     details: list[str] | dict[Any, Any] | str = ""
@@ -49,7 +51,7 @@ def run_command(command: Command) -> None:
         command.errors = [str(exc.stderr)]
 
 
-def worker(pending_queue: Queue, completed_queue: Queue) -> None:
+def worker(pending_queue: Queue[Any], completed_queue: Queue[Any]) -> None:
     """Run a command from pending, parse, and place in completed.
 
     :param pending_queue: A queue with plugins to process
@@ -75,8 +77,8 @@ class CommandRunner:
 
     def __init__(self):
         """Initialize the command runner."""
-        self._completed_queue: Queue | None = None
-        self._pending_queue: Queue | None = None
+        self._completed_queue: Queue[Any] | None = None
+        self._pending_queue: Queue[Any] | None = None
 
     def run_multi_thread(self, command_classes):
         """Run commands with multiple threads.

src/ansible_navigator/image_manager/inspector.py

@@ -89,7 +89,7 @@ def parse(command: Command):
             command.details = valid_images
 
 
-def inspect_all(container_engine: str) -> tuple[list, str]:
+def inspect_all(container_engine: str) -> tuple[list[str], str]:
     """Run inspect against all images in the list.
 
     :param container_engine: Name of the container engine

src/ansible_navigator/runner/ansible_doc.py

@@ -15,7 +15,7 @@ class AnsibleDoc(Base):
 
     def fetch_plugin_doc(
         self,
-        plugin_names: list,
+        plugin_names: list[str | None],
         plugin_type: str | None = None,
         response_format: str | None = "json",
         snippet: bool | None = None,

src/ansible_navigator/runner/ansible_inventory.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from typing import Any
+
 from ansible_runner import get_inventory
 
 from .base import Base
@@ -14,7 +16,7 @@ class AnsibleInventory(Base):
     def fetch_inventory(
         self,
         action: str,
-        inventories: list,
+        inventories: list[Any],
         response_format: str | None = None,
         host: str | None = None,
         playbook_dir: str | None = None,

src/ansible_navigator/runner/base.py

@@ -85,7 +85,7 @@ def __init__(
         self.cancelled: bool = False
         self.finished: bool = False
         self.status: str | None = None
-        self._runner_args: dict = {}
+        self._runner_args: dict[str, Any] = {}
 
         # when the ce is podman, set the container user to root
         if self._ce == "podman":

src/ansible_navigator/runner/command_async.py

@@ -8,6 +8,7 @@
 
 from copy import deepcopy
 from queue import Queue
+from typing import Any
 
 from ansible_runner import run_command_async
 
@@ -17,7 +18,7 @@
 class CommandAsync(CommandBase):
     """A wrapper for the asynchronous runner."""
 
-    def __init__(self, executable_cmd: str, queue: Queue, write_job_events: bool, **kwargs):
+    def __init__(self, executable_cmd: str, queue: Queue[Any], write_job_events: bool, **kwargs):
         """Initialize the arguments for the ``run_command_async`` interface of ``ansible-runner``.
 
         For common arguments refer to the documentation of the ``CommandBase`` class.

src/ansible_navigator/runner/command_base.py

@@ -16,9 +16,9 @@ class CommandBase(Base):
     def __init__(
         self,
         executable_cmd: str,
-        cmdline: list | None = None,
+        cmdline: list[str] | None = None,
         playbook: str | None = None,
-        inventory: list | None = None,
+        inventory: list[str] | None = None,
         **kwargs,
     ):
         """Handle common arguments of ``run_command`` interface for ``ansible-runner``.

src/ansible_navigator/ui_framework/colorize.py

@@ -12,6 +12,7 @@
 import re
 
 from itertools import chain
+from typing import Any
 
 from ansible_navigator.tm_tokenize.grammars import Grammars
 from ansible_navigator.tm_tokenize.region import Regions
@@ -43,7 +44,7 @@
 class ColorSchema:
     """A storage mechanism for the schema (theme)."""
 
-    def __init__(self, schema: dict[str, str | list | dict]):
+    def __init__(self, schema: dict[str, str | list[Any] | dict[Any, Any]]):
         """Initialize the ColorSchema class.
 
         :param schema: The color scheme, theme to use
@@ -164,7 +165,7 @@ def render(self, doc: str, scope: str) -> list[list[SimpleLinePart]]:
         return res
 
 
-def scope_to_list(scope: str | list) -> list:
+def scope_to_list(scope: str | list[Any]) -> list[Any]:
     """Convert a token scope to a list if necessary.
 
     A scope in a theme should always be a string or list,