From 5a7a879a2ca1a9803dd299cf6a31309dac5fc19d Mon Sep 17 00:00:00 2001 From: Leandro Damascena Date: Tue, 15 Apr 2025 22:58:16 +0100 Subject: [PATCH] Using generics types --- .../utilities/parameters/base.py | 3 +- .../_boto3/test_utilities_parameters.py | 84 ++++++++++--------- 2 files changed, 45 insertions(+), 42 deletions(-) diff --git a/aws_lambda_powertools/utilities/parameters/base.py b/aws_lambda_powertools/utilities/parameters/base.py index 610530f3263..42c15e11304 100644 --- a/aws_lambda_powertools/utilities/parameters/base.py +++ b/aws_lambda_powertools/utilities/parameters/base.py @@ -8,8 +8,9 @@ import os from abc import ABC, abstractmethod +from collections.abc import Callable from datetime import datetime, timedelta -from typing import TYPE_CHECKING, Any, Callable, NamedTuple, cast, overload +from typing import TYPE_CHECKING, Any, NamedTuple, cast, overload from aws_lambda_powertools.shared import constants, user_agent from aws_lambda_powertools.shared.functions import resolve_max_age diff --git a/tests/functional/parameters/_boto3/test_utilities_parameters.py b/tests/functional/parameters/_boto3/test_utilities_parameters.py index 42f654cbd1e..f7b7a642e00 100644 --- a/tests/functional/parameters/_boto3/test_utilities_parameters.py +++ b/tests/functional/parameters/_boto3/test_utilities_parameters.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import base64 import json import random @@ -5,7 +7,7 @@ import uuid from datetime import datetime, timedelta from io import BytesIO -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any import boto3 import pytest @@ -52,9 +54,9 @@ def mock_binary_value() -> str: def build_get_parameters_stub( - params: Dict[str, Any], - invalid_parameters: Optional[List[str]] = None, -) -> Dict[str, List]: + params: dict[str, Any], + invalid_parameters: list[str] | None = None, +) -> dict[str, list]: invalid_parameters = invalid_parameters or [] version = random.randrange(1, 1000) return { @@ -527,7 +529,7 @@ def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs) -> st def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "ssm", TestProvider()) @@ -685,7 +687,7 @@ def set(self, name: str, value: Any, *, overwrite: bool = False, **kwargs) -> st def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "secrets", TestProvider()) @@ -1025,7 +1027,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: return mock_value - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: ... + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: ... monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "ssm", TestProvider()) monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "secrets", TestProvider()) @@ -1875,7 +1877,7 @@ def _get(self, name: str, **kwargs) -> str: assert name == mock_name raise Exception("test exception raised") - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() provider = TestProvider() @@ -1895,7 +1897,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name raise Exception("test exception raised") @@ -1919,7 +1921,7 @@ def _get(self, name: str, **kwargs) -> str: assert name == mock_name return mock_data - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() provider = TestProvider() @@ -1943,7 +1945,7 @@ def _get(self, name: str, **kwargs) -> str: assert name == mock_name return mock_data - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() provider = TestProvider() @@ -1967,7 +1969,7 @@ def _get(self, name: str, **kwargs) -> str: assert name == mock_name return mock_data - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() provider = TestProvider() @@ -1991,7 +1993,7 @@ def _get(self, name: str, **kwargs) -> str: assert name == mock_name return mock_data - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() provider = TestProvider() @@ -2013,7 +2015,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name return {"A": mock_data} @@ -2036,7 +2038,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name return {"A": mock_data, "B": mock_data + "{"} @@ -2060,7 +2062,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name return {"A": mock_data} @@ -2084,7 +2086,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name return {"A": mock_data} @@ -2109,7 +2111,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name return {"A": mock_data_a, "B": mock_data_b} @@ -2133,7 +2135,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name return {"A": mock_data} @@ -2154,7 +2156,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() provider = TestProvider() @@ -2177,7 +2179,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name return {"A": mock_value} @@ -2201,7 +2203,7 @@ def _get(self, name: str, **kwargs) -> str: assert name == mock_name return mock_value - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "ssm", TestProvider()) @@ -2218,7 +2220,7 @@ class TestProvider(SSMProvider): def __init__(self, boto_config: Config = config, **kwargs): super().__init__(boto_config=boto_config, **kwargs) - def get_parameters_by_name(self, *args, **kwargs) -> Union[Dict[str, str], Dict[str, bytes], Dict[str, dict]]: + def get_parameters_by_name(self, *args, **kwargs) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: return {mock_name: mock_value} monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "ssm", TestProvider()) @@ -2247,7 +2249,7 @@ def _get(self, name: str, decrypt: bool = False, **sdk_options) -> str: assert decrypt return decrypted_response - def _get_parameters_by_name(self, *args, **kwargs) -> Tuple[Dict[str, Any], List[str]]: + def _get_parameters_by_name(self, *args, **kwargs) -> tuple[dict[str, Any], list[str]]: return {mock_name: mock_value}, [] monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "ssm", TestProvider()) @@ -2276,10 +2278,10 @@ def __init__(self, boto_config: Config = config, **kwargs): # def _get_parameters_by_name(self, parameters: Dict[str, Dict], raise_on_error: bool = True) -> Dict[str, Any]: def _get_parameters_by_name( self, - parameters: Dict[str, Dict], + parameters: dict[str, dict], raise_on_error: bool = True, decrypt: bool = False, - ) -> Tuple[Dict[str, Any], List[str]]: + ) -> tuple[dict[str, Any], list[str]]: # THEN max_age should use no_cache_param override assert parameters[mock_name]["max_age"] == 0 assert parameters["no-override"]["max_age"] == default_cache_period @@ -2302,10 +2304,10 @@ def __init__(self, boto_config: Config = config, **kwargs): def _get_parameters_by_name( self, - parameters: Dict[str, Dict], + parameters: dict[str, dict], raise_on_error: bool = True, decrypt: bool = False, - ) -> Tuple[Dict[str, Any], List[str]]: + ) -> tuple[dict[str, Any], list[str]]: # THEN we should always split to respect GetParameters max assert len(parameters) == self._MAX_GET_PARAMETERS_ITEM return {}, [] @@ -2325,7 +2327,7 @@ class TestProvider(SSMProvider): def __init__(self, boto_config: Config = config, **kwargs): super().__init__(boto_config=boto_config, **kwargs) - def _get_parameters_by_name(self, *args, **kwargs) -> Tuple[Dict[str, Any], List[str]]: + def _get_parameters_by_name(self, *args, **kwargs) -> tuple[dict[str, Any], list[str]]: raise RuntimeError("Should not be called if it's in cache") provider = TestProvider() @@ -2389,7 +2391,7 @@ def _get(self, name: str, **kwargs) -> str: assert not kwargs["decrypt"] return mock_value - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() monkeypatch.setattr(parameters.ssm, "DEFAULT_PROVIDERS", {}) @@ -2409,7 +2411,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name return {"A": mock_value} @@ -2430,7 +2432,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name assert kwargs["recursive"] assert not kwargs["decrypt"] @@ -2454,7 +2456,7 @@ class TestProvider(SSMProvider): def __init__(self, boto_config: Config = config, **kwargs): super().__init__(boto_config=boto_config, **kwargs) - def get_parameters_by_name(self, *args, **kwargs) -> Union[Dict[str, str], Dict[str, bytes], Dict[str, dict]]: + def get_parameters_by_name(self, *args, **kwargs) -> dict[str, str] | dict[str, bytes] | dict[str, dict]: return {mock_name: mock_value} monkeypatch.setattr(parameters.ssm, "DEFAULT_PROVIDERS", {}) @@ -2475,7 +2477,7 @@ def _get(self, name: str, **kwargs) -> str: assert name == mock_name return mock_value - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "secrets", TestProvider()) @@ -2495,7 +2497,7 @@ def _get(self, name: str, **kwargs) -> str: assert name == mock_name return mock_value - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() monkeypatch.setattr(parameters.secrets, "DEFAULT_PROVIDERS", {}) @@ -2689,7 +2691,7 @@ def _get(self, name: str, **kwargs) -> bytes: assert name == mock_name return mock_body_bytes - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "appconfig", TestProvider()) @@ -2714,7 +2716,7 @@ def _get(self, name: str, **kwargs) -> str: assert name == mock_name return mock_body_bytes - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() monkeypatch.setitem(parameters.base.DEFAULT_PROVIDERS, "appconfig", TestProvider()) @@ -2737,7 +2739,7 @@ def get(self, name: str, **kwargs) -> str: def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() monkeypatch.setattr(parameters.appconfig, "DEFAULT_PROVIDERS", {}) @@ -2877,7 +2879,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise NotImplementedError() - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: assert path == mock_name return {"A": mock_value} @@ -2900,7 +2902,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: return mock_value - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: raise NotImplementedError() provider = TestProvider() @@ -2934,7 +2936,7 @@ class TestProvider(BaseProvider): def _get(self, name: str, **kwargs) -> str: raise ValueError("This parameter doesn't exist") - def _get_multiple(self, path: str, **kwargs) -> Dict[str, str]: + def _get_multiple(self, path: str, **kwargs) -> dict[str, str]: return {"A": mock_value} provider = TestProvider()