Skip to content

refactor(parameters): use standard collections for types #6481

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Apr 16, 2025
Merged

refactor(parameters): use standard collections for types #6481

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5a7a879
Using generics types
leandrodamascena Apr 15, 2025
fc92361
Merge branch 'develop' into refactor/migrate-callable-parameters
anafalcao Apr 16, 2025
99a6491
Merge branch 'develop' into refactor/migrate-callable-parameters
leandrodamascena Apr 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion aws_lambda_powertools/utilities/parameters/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
84 changes: 43 additions & 41 deletions tests/functional/parameters/_boto3/test_utilities_parameters.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
from __future__ import annotations

import base64
import json
import random
import string
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
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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())
Expand Down Expand Up @@ -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())
Expand Down Expand Up @@ -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())
Expand Down Expand Up @@ -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()
Expand All @@ -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")

Expand All @@ -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()
Expand All @@ -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()
Expand All @@ -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()
Expand All @@ -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()
Expand All @@ -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}

Expand All @@ -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 + "{"}

Expand All @@ -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}

Expand All @@ -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}

Expand All @@ -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}

Expand All @@ -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}

Expand All @@ -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()
Expand All @@ -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}

Expand All @@ -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())
Expand All @@ -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())
Expand Down Expand Up @@ -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())
Expand Down Expand Up @@ -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
Expand All @@ -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 {}, []
Expand All @@ -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()
Expand Down Expand Up @@ -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", {})
Expand All @@ -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}

Expand All @@ -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"]
Expand All @@ -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", {})
Expand All @@ -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())
Expand All @@ -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", {})
Expand Down Expand Up @@ -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())
Expand All @@ -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())
Expand All @@ -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", {})
Expand Down Expand Up @@ -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}

Expand All @@ -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()
Expand Down Expand Up @@ -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()
Expand Down
Loading