Skip to content

refactor(data_masking): add from __future__ import annotations #4945

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.

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 1 commit into from
Aug 15, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
18 changes: 10 additions & 8 deletions aws_lambda_powertools/utilities/data_masking/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,7 @@
import functools
import logging
import warnings
from numbers import Number
from typing import Any, Callable, Mapping, Optional, Sequence, Union, overload
from typing import TYPE_CHECKING, Any, Callable, Mapping, Sequence, overload

from jsonpath_ng.ext import parse

Expand All @@ -14,6 +13,9 @@
)
from aws_lambda_powertools.utilities.data_masking.provider import BaseProvider

if TYPE_CHECKING:
from numbers import Number

logger = logging.getLogger(__name__)


Expand Down Expand Up @@ -43,7 +45,7 @@ def lambda_handler(event, context):

def __init__(
self,
provider: Optional[BaseProvider] = None,
provider: BaseProvider | None = None,
raise_on_missing_field: bool = True,
):
self.provider = provider or BaseProvider()
Expand Down Expand Up @@ -111,7 +113,7 @@ def _apply_action(
----------
data : str | dict
The input data to process.
fields : Optional[List[str]]
fields : list[str] | None
A list of fields to apply the action to. If 'None', the action is applied to the entire 'data'.
action : Callable
The action to apply to the data. It should be a callable that performs an operation on the data
Expand Down Expand Up @@ -142,21 +144,21 @@ def _apply_action(

def _apply_action_to_fields(
self,
data: Union[dict, str],
data: dict | str,
fields: list,
action: Callable,
provider_options: dict | None = None,
**encryption_context: str,
) -> Union[dict, str]:
) -> dict | str:
"""
This method takes the input data, which can be either a dictionary or a JSON string,
and erases, encrypts, or decrypts the specified fields.

Parameters
----------
data : Union[dict, str])
data : dict | str)
The input data to process. It can be either a dictionary or a JSON string.
fields : List
fields : list
A list of fields to apply the action to. Each field can be specified as a string or
a list of strings representing nested keys in the dictionary.
action : Callable
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ def encrypt(self, data) -> str:
def decrypt(self, data) -> Any:
# Implementation logic for data decryption

def erase(self, data) -> Union[str, Iterable]:
def erase(self, data) -> str | Iterable:
# Implementation logic for data masking
pass

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import json
import logging
from binascii import Error
from typing import Any, Callable, List
from typing import Any, Callable

import botocore
from aws_encryption_sdk import (
Expand All @@ -18,7 +18,6 @@
GenerateKeyError,
NotSupportedError,
)
from aws_encryption_sdk.structures import MessageHeader

from aws_lambda_powertools.shared.functions import (
base64_decode,
Expand Down Expand Up @@ -76,7 +75,7 @@ def lambda_handler(event, context):

def __init__(
self,
keys: List[str],
keys: list[str],
key_provider=None,
local_cache_capacity: int = CACHE_CAPACITY,
max_cache_age_seconds: float = MAX_CACHE_AGE_SECONDS,
Expand Down Expand Up @@ -112,7 +111,7 @@ class KMSKeyProvider:

def __init__(
self,
keys: List[str],
keys: list[str],
json_serializer: Callable[..., str],
json_deserializer: Callable[[str], Any],
local_cache_capacity: int = CACHE_CAPACITY,
Expand Down Expand Up @@ -143,7 +142,7 @@ def encrypt(self, data: Any, provider_options: dict | None = None, **encryption_

Parameters
-------
data : Union[bytes, str]
data : Any
The data to be encrypted.
provider_options : dict
Additional options for the aws_encryption_sdk.EncryptionSDKClient
Expand Down Expand Up @@ -180,7 +179,7 @@ def decrypt(self, data: str, provider_options: dict | None = None, **encryption_

Parameters
-------
data : Union[bytes, str]
data : str
The encrypted data, as a base64-encoded string
provider_options
Additional options for the aws_encryption_sdk.EncryptionSDKClient
Expand All @@ -201,8 +200,6 @@ def decrypt(self, data: str, provider_options: dict | None = None, **encryption_
)

try:
decryptor_header: MessageHeader

ciphertext, decryptor_header = self.client.decrypt(
source=ciphertext_decoded,
key_provider=self.key_provider,
Expand Down
Loading