From f747ed0cddf679561f09d2116859d2416e8f5864 Mon Sep 17 00:00:00 2001 From: Mark Chappell Date: Sun, 2 Feb 2025 15:19:35 +0100 Subject: [PATCH] Minor typing fixups --- changelogs/fragments/20250202-typealias.yml | 7 ++++ plugins/module_utils/_s3/common.py | 3 +- plugins/module_utils/botocore.py | 40 ++++++++++----------- plugins/module_utils/exceptions.py | 4 +-- plugins/plugin_utils/botocore.py | 5 +-- 5 files changed, 32 insertions(+), 27 deletions(-) create mode 100644 changelogs/fragments/20250202-typealias.yml diff --git a/changelogs/fragments/20250202-typealias.yml b/changelogs/fragments/20250202-typealias.yml new file mode 100644 index 00000000000..a804ab18ee4 --- /dev/null +++ b/changelogs/fragments/20250202-typealias.yml @@ -0,0 +1,7 @@ +bugfixes: + - module_utils.botocore - fixed type aliasing(). + - plugin_utils.botocore - fixed type aliasing(). +minor_changes: + - module_utils._s3 - explicitly cast super to the parent type(). + - module_utils.botocore - avoid assigning unused parts of exc_info return(). + - module_utils.exceptions - avoid assigning unused parts of exc_info return(). diff --git a/plugins/module_utils/_s3/common.py b/plugins/module_utils/_s3/common.py index 46049c15449..a6b7d1f668f 100644 --- a/plugins/module_utils/_s3/common.py +++ b/plugins/module_utils/_s3/common.py @@ -6,6 +6,7 @@ from __future__ import annotations import functools +from typing import cast try: # Beware, S3 is a "special" case, it sometimes catches botocore exceptions and @@ -67,7 +68,7 @@ def _is_missing(cls): @classmethod def common_error_handler(cls, description): def wrapper(func): - parent_class = super(S3ErrorHandler, cls) + parent_class = cast(AWSErrorHandler, super(S3ErrorHandler, cls)) @parent_class.common_error_handler(description) @functools.wraps(func) diff --git a/plugins/module_utils/botocore.py b/plugins/module_utils/botocore.py index 5970b1e32db..64ad6598fc6 100644 --- a/plugins/module_utils/botocore.py +++ b/plugins/module_utils/botocore.py @@ -37,6 +37,7 @@ import json import os +import sys import traceback import typing from typing import Any @@ -48,11 +49,13 @@ from typing import Tuple from typing import Type from typing import Union +from typing_extensions import TypeAlias BOTO3_IMP_ERR = None try: import boto3 import botocore + from botocore import ClientError HAS_BOTO3 = True except ImportError: @@ -60,8 +63,8 @@ HAS_BOTO3 = False if typing.TYPE_CHECKING: - ClientType = botocore.client.BaseClient - ResourceType = boto3.resources.base.ServiceResource + ClientType: TypeAlias = botocore.client.BaseClient + ResourceType: TypeAlias = boto3.resources.base.ServiceResource BotoConn = Union[ClientType, ResourceType, Tuple[ClientType, ResourceType]] from .modules import AnsibleAWSModule @@ -114,6 +117,7 @@ def boto3_conn( return _boto3_conn(conn_type=conn_type, resource=resource, region=region, endpoint=endpoint, **params) except botocore.exceptions.NoRegionError: module.fail_json( + # pylint: disable-next=protected-access msg=f"The {module._name} module requires a region and none was found in configuration, " "environment variables or module parameters", ) @@ -228,7 +232,7 @@ def _aws_region(params: Dict) -> Optional[str]: def get_aws_region( module: AnsibleAWSModule, -) -> Optional[str]: # pylint: disable=redefined-outer-name +) -> Optional[str]: try: return _aws_region(module.params) except AnsibleBotocoreError as e: @@ -294,7 +298,7 @@ def _aws_connection_info(params: Dict) -> Tuple[Optional[str], Optional[str], Di def get_aws_connection_info( module: AnsibleAWSModule, -) -> Tuple[Optional[str], Optional[str], Dict]: # pylint: disable=redefined-outer-name +) -> Tuple[Optional[str], Optional[str], Dict]: try: return _aws_connection_info(module.params) except AnsibleBotocoreError as e: @@ -369,12 +373,8 @@ def is_boto3_error_code( except botocore.exceptions.ClientError as e: # handle the generic error case for all other codes """ - from botocore.exceptions import ClientError - if e is None: - import sys - - dummy, e, dummy = sys.exc_info() + e = sys.exc_info()[1] if not isinstance(code, (list, tuple, set)): code = [code] if isinstance(e, ClientError) and e.response["Error"]["Code"] in code: @@ -398,12 +398,9 @@ def is_boto3_error_message( except botocore.exceptions.ClientError as e: # handle the generic error case for all other codes """ - from botocore.exceptions import ClientError if e is None: - import sys - - dummy, e, dummy = sys.exc_info() + e = sys.exc_info()[1] if isinstance(e, ClientError) and msg in e.response["Error"]["Message"]: return ClientError return type("NeverEverRaisedException", (Exception,), {}) @@ -415,7 +412,7 @@ def get_boto3_client_method_parameters( required=False, ) -> List: op = client.meta.method_to_api_mapping.get(method_name) - input_shape = client._service_model.operation_model(op).input_shape + input_shape = client._service_model.operation_model(op).input_shape # pylint: disable=protected-access if not input_shape: parameters = [] elif required: @@ -450,32 +447,31 @@ def enable_placebo(session: boto3.session.Session) -> None: """ Helper to record or replay offline modules for testing purpose. """ + # pylint: disable=import-outside-toplevel if "_ANSIBLE_PLACEBO_RECORD" in os.environ: import placebo - existing_entries = os.listdir(os.environ["_ANSIBLE_PLACEBO_RECORD"]) - idx = len(existing_entries) - data_path = f"{os.environ['_ANSIBLE_PLACEBO_RECORD']}/{idx}" + recorded_entries = os.listdir(os.environ["_ANSIBLE_PLACEBO_RECORD"]) + data_path = f"{os.environ['_ANSIBLE_PLACEBO_RECORD']}/{len(recorded_entries)}" os.mkdir(data_path) pill = placebo.attach(session, data_path=data_path) pill.record() if "_ANSIBLE_PLACEBO_REPLAY" in os.environ: import shutil - import placebo - existing_entries = sorted([int(i) for i in os.listdir(os.environ["_ANSIBLE_PLACEBO_REPLAY"])]) - idx = str(existing_entries[0]) - data_path = os.environ["_ANSIBLE_PLACEBO_REPLAY"] + "/" + idx + replay_entries = sorted([int(i) for i in os.listdir(os.environ["_ANSIBLE_PLACEBO_REPLAY"])]) + data_path = f"{os.environ['_ANSIBLE_PLACEBO_REPLAY']}/{int(replay_entries[0])}" try: shutil.rmtree("_tmp") except FileNotFoundError: pass shutil.move(data_path, "_tmp") - if len(existing_entries) == 1: + if len(replay_entries) == 1: os.rmdir(os.environ["_ANSIBLE_PLACEBO_REPLAY"]) pill = placebo.attach(session, data_path="_tmp") pill.playback() + # pylint: enable=import-outside-toplevel def check_sdk_version_supported( diff --git a/plugins/module_utils/exceptions.py b/plugins/module_utils/exceptions.py index 3db5ae3d793..f44c0b547c6 100644 --- a/plugins/module_utils/exceptions.py +++ b/plugins/module_utils/exceptions.py @@ -48,7 +48,7 @@ def is_ansible_aws_error_code(code, e=None): if e is None: import sys - dummy, e, dummy = sys.exc_info() + e = sys.exc_info()[1] if not isinstance(code, (list, tuple, set)): code = [code] if ( @@ -79,7 +79,7 @@ def is_ansible_aws_error_message(msg, e=None): if e is None: import sys - dummy, e, dummy = sys.exc_info() + e = sys.exc_info()[1] if ( isinstance(e, AnsibleAWSError) and e.exception diff --git a/plugins/plugin_utils/botocore.py b/plugins/plugin_utils/botocore.py index b13c263ff28..f64bd436242 100644 --- a/plugins/plugin_utils/botocore.py +++ b/plugins/plugin_utils/botocore.py @@ -18,12 +18,13 @@ from typing import Optional from typing import Tuple from typing import Union +from typing_extensions import TypeAlias if typing.TYPE_CHECKING: from .base import AWSPluginBase - ClientType = botocore.client.BaseClient - ResourceType = boto3.resources.base.ServiceResource + ClientType: TypeAlias = botocore.client.BaseClient + ResourceType: TypeAlias = boto3.resources.base.ServiceResource BotoConn = Union[ClientType, ResourceType, Tuple[ClientType, ResourceType]] from ansible.module_utils.basic import to_native