Fix typing errors

Author: Shrews

src/ansible_runner/__main__.py

@@ -44,7 +44,8 @@
 from ansible_runner import run
 from ansible_runner import output
 from ansible_runner import cleanup
-from ansible_runner.utils import dump_artifact, Bunch, register_for_cleanup
+from ansible_runner._internal._dump_artifacts import dump_artifact
+from ansible_runner.utils import Bunch, register_for_cleanup
 from ansible_runner.utils.capacity import get_cpu_count, get_mem_in_bytes, ensure_uuid
 from ansible_runner.utils.importlib_compat import importlib_metadata
 from ansible_runner.runner import Runner

src/ansible_runner/_internal/_dump_artifacts.py

@@ -8,20 +8,23 @@
 import tempfile
 
 from collections.abc import MutableMapping
+from typing import Any
 
 from ansible_runner.config.runner import RunnerConfig
 from ansible_runner.utils import isinventory, isplaybook
 
 
 def dump_artifacts(config: RunnerConfig) -> None:
     """Introspect the arguments and dump objects to disk"""
+    private_data_dir = config.private_data_dir or ""
+
     if config.role:
-        role = {'name': config.role}
+        role: dict[str, Any] = {'name': config.role}
         if config.role_vars:
             role['vars'] = config.role_vars
 
         hosts = config.host_pattern or 'all'
-        play = [{'hosts': hosts, 'roles': [role]}]
+        play: list[dict[str, Any]] = [{'hosts': hosts, 'roles': [role]}]
 
         if config.role_skip_facts:
             play[0]['gather_facts'] = False
@@ -33,9 +36,9 @@ def dump_artifacts(config: RunnerConfig) -> None:
 
         roles_path = config.roles_path
         if not roles_path:
-            roles_path = os.path.join(config.private_data_dir, 'roles')
+            roles_path = os.path.join(private_data_dir, 'roles')
         else:
-            roles_path += f":{os.path.join(config.private_data_dir, 'roles')}"
+            roles_path += f":{os.path.join(private_data_dir, 'roles')}"
 
         config.envvars['ANSIBLE_ROLES_PATH'] = roles_path
 
@@ -46,12 +49,12 @@ def dump_artifacts(config: RunnerConfig) -> None:
             playbook = [playbook]
 
         if isplaybook(playbook):
-            path = os.path.join(config.private_data_dir, 'project')
+            path = os.path.join(private_data_dir, 'project')
             config.playbook = dump_artifact(json.dumps(playbook), path, 'main.json')
 
     obj = config.inventory
     if obj and isinventory(obj):
-        path = os.path.join(config.private_data_dir, 'inventory')
+        path = os.path.join(private_data_dir, 'inventory')
         if isinstance(obj, MutableMapping):
             config.inventory = dump_artifact(json.dumps(obj), path, 'hosts.json')
         elif isinstance(obj, str):
@@ -65,14 +68,14 @@ def dump_artifacts(config: RunnerConfig) -> None:
     if not config.suppress_env_files:
         for key in ('envvars', 'extravars', 'passwords', 'settings'):
             obj = getattr(config, key, None)
-            if obj and not os.path.exists(os.path.join(config.private_data_dir, 'env', key)):
-                path = os.path.join(config.private_data_dir, 'env')
+            if obj and not os.path.exists(os.path.join(private_data_dir, 'env', key)):
+                path = os.path.join(private_data_dir, 'env')
                 dump_artifact(json.dumps(obj), path, key)
 
         for key in ('ssh_key', 'cmdline'):
             obj = getattr(config, key, None)
-            if obj and not os.path.exists(os.path.join(config.private_data_dir, 'env', key)):
-                path = os.path.join(config.private_data_dir, 'env')
+            if obj and not os.path.exists(os.path.join(private_data_dir, 'env', key)):
+                path = os.path.join(private_data_dir, 'env')
                 dump_artifact(obj, path, key)
 
 

src/ansible_runner/interface.py

@@ -67,19 +67,20 @@ def init_runner(
 
     if streamer:
         # undo any full paths that were dumped by dump_artifacts above in the streamer case
-        private_data_dir = config.private_data_dir
+        private_data_dir = config.private_data_dir or ""
         project_dir = os.path.join(private_data_dir, 'project')
 
         playbook_path = config.playbook or ''
-        if os.path.isabs(playbook_path) and playbook_path.startswith(project_dir):
+        if isinstance(playbook_path, str) and os.path.isabs(playbook_path) and playbook_path.startswith(project_dir):
             config.playbook = os.path.relpath(playbook_path, project_dir)
 
         inventory_path = config.inventory or ''
-        if os.path.isabs(inventory_path) and inventory_path.startswith(private_data_dir):
+        if isinstance(inventory_path, str) and os.path.isabs(inventory_path) and inventory_path.startswith(private_data_dir):
             config.inventory = os.path.relpath(inventory_path, private_data_dir)
 
-        envvars = config.envvars or {}
-        roles_path = envvars.get('ANSIBLE_ROLES_PATH') or ''
+        if config.envvars is None:
+            config.envvars = {}
+        roles_path = config.envvars.get('ANSIBLE_ROLES_PATH') or ''
         if os.path.isabs(roles_path) and roles_path.startswith(private_data_dir):
             config.envvars['ANSIBLE_ROLES_PATH'] = os.path.relpath(roles_path, private_data_dir)
 

src/ansible_runner/streaming.py

@@ -1,7 +1,6 @@
 from __future__ import annotations  # allow newer type syntax until 3.10 is our minimum
 
 import codecs
-import io
 import json
 import os
 import stat
@@ -13,6 +12,7 @@
 from collections.abc import Mapping
 from functools import wraps
 from threading import Event, RLock, Thread
+from typing import BinaryIO
 
 import ansible_runner
 from ansible_runner.exceptions import ConfigurationError
@@ -38,7 +38,7 @@ def __init__(self, settings):
 
 
 class Transmitter:
-    def __init__(self, only_transmit_kwargs: bool, _output: io.FileIO | None, **kwargs):
+    def __init__(self, only_transmit_kwargs: bool, _output: BinaryIO | None, **kwargs):
         if _output is None:
             _output = sys.stdout.buffer
         self._output = _output