docs(parameters): extract and fix code examples

Makefile

@@ -90,3 +90,10 @@ changelog:
 
 mypy:
 	poetry run mypy --pretty aws_lambda_powertools
+
+format-examples:
+	poetry run isort docs/examples
+	poetry run black docs/examples/*/*/*.py
+
+lint-examples:
+	poetry run python3 -m py_compile docs/examples/*/*/*.py

docs/examples/utilities/parameters/app_config_provider.py

@@ -0,0 +1,15 @@
+from botocore.config import Config
+
+from aws_lambda_powertools.utilities import parameters
+
+config = Config(region_name="us-west-1")
+appconf_provider = parameters.AppConfigProvider(
+    environment="my_env",
+    application="my_app",
+    config=config,
+)
+
+
+def handler(event, context):
+    # Retrieve a single secret
+    value: bytes = appconf_provider.get("my_conf")

docs/examples/utilities/parameters/create_your_own_s3_provider.py

@@ -0,0 +1,53 @@
+import copy
+from typing import Dict
+
+import boto3
+
+from aws_lambda_powertools.utilities import BaseProvider
+
+
+class S3Provider(BaseProvider):
+    bucket_name = None
+    client = None
+
+    def __init__(self, bucket_name: str):
+        # Initialize the client to your custom parameter store
+        # E.g.:
+
+        self.bucket_name = bucket_name
+        self.client = boto3.client("s3")
+
+    def _get(self, name: str, **sdk_options) -> str:
+        # Retrieve a single value
+        # E.g.:
+
+        sdk_options["Bucket"] = self.bucket_name
+        sdk_options["Key"] = name
+
+        response = self.client.get_object(**sdk_options)
+        return
+
+    def _get_multiple(self, path: str, **sdk_options) -> Dict[str, str]:
+        # Retrieve multiple values
+        # E.g.:
+
+        list_sdk_options = copy.deepcopy(sdk_options)
+
+        list_sdk_options["Bucket"] = self.bucket_name
+        list_sdk_options["Prefix"] = path
+
+        list_response = self.client.list_objects_v2(**list_sdk_options)
+
+        parameters = {}
+
+        for obj in list_response.get("Contents", []):
+            get_sdk_options = copy.deepcopy(sdk_options)
+
+            get_sdk_options["Bucket"] = self.bucket_name
+            get_sdk_options["Key"] = obj["Key"]
+
+            get_response = self.client.get_object(**get_sdk_options)
+
+            parameters[obj["Key"]] = get_response["Body"].read().decode()
+
+        return parameters

docs/examples/utilities/parameters/custom_caching_parameters.py

@@ -0,0 +1,16 @@
+from botocore.config import Config
+
+from aws_lambda_powertools.utilities import parameters
+
+config = Config(region_name="us-west-1")
+ssm_provider = parameters.SSMProvider(config=config)
+
+
+def handler(event, context):
+    # Retrieve a single parameter
+    value = ssm_provider.get("/my/parameter", max_age=60)  # 1 minute
+
+    # Retrieve multiple parameters from a path prefix
+    values = ssm_provider.get_multiple("/my/path/prefix", max_age=60)
+    for k, v in values.items():
+        print(f"{k}: {v}")

docs/examples/utilities/parameters/dynamodb_provider.py

@@ -0,0 +1,8 @@
+from aws_lambda_powertools.utilities import parameters
+
+dynamodb_provider = parameters.DynamoDBProvider(table_name="my-table")
+
+
+def handler(event, context):
+    # Retrieve a value from DynamoDB
+    value = dynamodb_provider.get("my-parameter")

docs/examples/utilities/parameters/dynamodb_provider_customization.py

@@ -0,0 +1,12 @@
+from aws_lambda_powertools.utilities import parameters
+
+dynamodb_provider = parameters.DynamoDBProvider(
+    table_name="my-table",
+    key_attr="MyKeyAttr",
+    sort_attr="MySortAttr",
+    value_attr="MyvalueAttr",
+)
+
+
+def handler(event, context):
+    value = dynamodb_provider.get("my-parameter")

docs/examples/utilities/parameters/dynamodb_provider_get_multiple.py

@@ -0,0 +1,13 @@
+from aws_lambda_powertools.utilities import parameters
+
+dynamodb_provider = parameters.DynamoDBProvider(table_name="my-table")
+
+
+def handler(event, context):
+    # Retrieve multiple values by performing a Query on the DynamoDB table
+    # This returns a dict with the sort key attribute as dict key.
+    parameters = dynamodb_provider.get_multiple("my-hash-key")
+    for k, v in parameters.items():
+        # k: param-a
+        # v: "my-value-a"
+        print(f"{k}: {v}")

docs/examples/utilities/parameters/dynamodb_provider_local.py

@@ -0,0 +1,6 @@
+from aws_lambda_powertools.utilities import parameters
+
+dynamodb_provider = parameters.DynamoDBProvider(
+    table_name="my-table",
+    endpoint_url="http://localhost:8000",
+)

docs/examples/utilities/parameters/fetching_app_config.py

@@ -0,0 +1,10 @@
+from aws_lambda_powertools.utilities import parameters
+
+
+def handler(event, context):
+    # Retrieve a single configuration, latest version
+    value: bytes = parameters.get_app_config(
+        name="my_configuration",
+        environment="my_env",
+        application="my_app",
+    )

docs/examples/utilities/parameters/fetching_secrets.py

@@ -0,0 +1,6 @@
+from aws_lambda_powertools.utilities import parameters
+
+
+def handler(event, context):
+    # Retrieve a single secret
+    value = parameters.get_secret("my-secret")

docs/examples/utilities/parameters/force_fetch_parameters.py

@@ -0,0 +1,6 @@
+from aws_lambda_powertools.utilities import parameters
+
+
+def handler(event, context):
+    # Retrieve a single parameter
+    value = parameters.get_parameter("/my/parameter", force_fetch=True)

docs/examples/utilities/parameters/parameters_boto3_client.py

@@ -0,0 +1,17 @@
+import boto3
+from botocore.config import Config
+
+from aws_lambda_powertools.utilities import parameters
+
+config = Config(region_name="us-west-1")
+
+# construct boto clients with any custom configuration
+ssm = boto3.client("ssm", config=config)
+secrets = boto3.client("secrets", config=config)
+appconfig = boto3.client("appconfig", config=config)
+dynamodb = boto3.resource("dynamodb", config=config)
+
+ssm_provider = parameters.SSMProvider(boto3_client=ssm)
+secrets_provider = parameters.SecretsProvider(boto3_client=secrets)
+appconf_provider = parameters.AppConfigProvider(boto3_client=appconfig, environment="my_env", application="my_app")
+dynamodb_provider = parameters.DynamoDBProvider(boto3_client=dynamodb, table_name="my-table")

docs/examples/utilities/parameters/parameters_custom_client.py

@@ -0,0 +1,12 @@
+import boto3
+
+from aws_lambda_powertools.utilities import parameters
+
+boto3_client = boto3.client("ssm")
+ssm_provider = parameters.SSMProvider(boto3_client=boto3_client)
+
+
+def handler(event, context):
+    # Retrieve a single parameter
+    value = ssm_provider.get("/my/parameter")
+    ...

docs/examples/utilities/parameters/parameters_custom_config.py

@@ -0,0 +1,12 @@
+from botocore.config import Config
+
+from aws_lambda_powertools.utilities import parameters
+
+boto_config = Config()
+ssm_provider = parameters.SSMProvider(config=boto_config)
+
+
+def handler(event, context):
+    # Retrieve a single parameter
+    value = ssm_provider.get("/my/parameter")
+    ...

docs/examples/utilities/parameters/parameters_custom_session.py

@@ -0,0 +1,12 @@
+import boto3
+
+from aws_lambda_powertools.utilities import parameters
+
+boto3_session = boto3.session.Session()
+ssm_provider = parameters.SSMProvider(boto3_session=boto3_session)
+
+
+def handler(event, context):
+    # Retrieve a single parameter
+    value = ssm_provider.get("/my/parameter")
+    ...

docs/examples/utilities/parameters/parameters_sdk_args.py

@@ -0,0 +1,8 @@
+from aws_lambda_powertools.utilities import parameters
+
+secrets_provider = parameters.SecretsProvider()
+
+
+def handler(event, context):
+    # The 'VersionId' argument will be passed to the underlying get_secret_value() call.
+    value = secrets_provider.get("my-secret", VersionId="e62ec170-6b01-48c7-94f3-d7497851a8d2")

docs/examples/utilities/parameters/parameters_transform.py

@@ -0,0 +1,5 @@
+from aws_lambda_powertools.utilities import parameters
+
+
+def handler(event, context):
+    value_from_json = parameters.get_parameter("/my/json/parameter", transform="json")

docs/examples/utilities/parameters/parameters_transform_auto.py

@@ -0,0 +1,7 @@
+from aws_lambda_powertools.utilities import parameters
+
+ssm_provider = parameters.SSMProvider()
+
+
+def handler(event, context):
+    values = ssm_provider.get_multiple("/param", transform="auto")

docs/examples/utilities/parameters/parameters_transform_providers.py

@@ -0,0 +1,11 @@
+from aws_lambda_powertools.utilities import parameters
+
+ssm_provider = parameters.SSMProvider()
+
+
+def handler(event, context):
+    # Transform a JSON string
+    value_from_json = ssm_provider.get("/my/json/parameter", transform="json")
+
+    # Transform a Base64 encoded string
+    value_from_binary = ssm_provider.get("/my/binary/parameter", transform="binary")

docs/examples/utilities/parameters/parameters_transform_raise_on_transform_error.py

@@ -0,0 +1,19 @@
+from aws_lambda_powertools.utilities import parameters
+
+ssm_provider = parameters.SSMProvider()
+
+
+def handler(event, context):
+    # This will display:
+    # /param/a: [some value]
+    # /param/b: [some value]
+    # /param/c: None
+    values = ssm_provider.get_multiple("/param", transform="json")
+    for k, v in values.items():
+        print(f"{k}: {v}")
+
+    try:
+        # This will raise a TransformParameterError exception
+        values = ssm_provider.get_multiple("/param", transform="json", raise_on_transform_error=True)
+    except parameters.exceptions.TransformParameterError:
+        ...

docs/examples/utilities/parameters/recursively_parameters.py

@@ -0,0 +1,12 @@
+from aws_lambda_powertools.utilities import parameters
+
+
+def handler(event, context):
+    # Retrieve a single parameter
+    value = parameters.get_parameter("/my/parameter")
+
+    # Retrieve multiple parameters from a path prefix recursively
+    # This returns a dict with the parameter name as key
+    values = parameters.get_parameters("/my/path/prefix")
+    for k, v in values.items():
+        print(f"{k}: {v}")

docs/examples/utilities/parameters/secrets_provider.py

@@ -0,0 +1,11 @@
+from botocore.config import Config
+
+from aws_lambda_powertools.utilities import parameters
+
+config = Config(region_name="us-west-1")
+secrets_provider = parameters.SecretsProvider(config=config)
+
+
+def handler(event, context):
+    # Retrieve a single secret
+    value = secrets_provider.get("my-secret")

docs/examples/utilities/parameters/ssm_provider.py

@@ -0,0 +1,16 @@
+from botocore.config import Config
+
+from aws_lambda_powertools.utilities import parameters
+
+config = Config(region_name="us-west-1")
+ssm_provider = parameters.SSMProvider(config=config)  # or boto3_session=boto3.Session()
+
+
+def handler(event, context):
+    # Retrieve a single parameter
+    value = ssm_provider.get("/my/parameter")
+
+    # Retrieve multiple parameters from a path prefix
+    values = ssm_provider.get_multiple("/my/path/prefix")
+    for k, v in values.items():
+        print(f"{k}: {v}")

docs/examples/utilities/parameters/ssm_provider_get_options.py

@@ -0,0 +1,9 @@
+from aws_lambda_powertools.utilities import parameters
+
+ssm_provider = parameters.SSMProvider()
+
+
+def handler(event, context):
+    decrypted_value = ssm_provider.get("/my/encrypted/parameter", decrypt=True)
+
+    no_recursive_values = ssm_provider.get_multiple("/my/path/prefix", recursive=False)

docs/examples/utilities/parameters/testing_parameters_app.py

@@ -0,0 +1,10 @@
+from botocore.config import Config
+
+from aws_lambda_powertools.utilities import parameters
+
+ssm_provider = parameters.SSMProvider(config=Config(region_name="us-west-1"))
+
+
+def handler(event, context):
+    value = ssm_provider.get("/my/parameter")
+    return {"message": value}

docs/examples/utilities/parameters/testing_parameters_clear_cache.py

@@ -0,0 +1,22 @@
+import pytest
+from src import app
+
+
+@pytest.fixture(scope="function", autouse=True)
+def clear_parameters_cache():
+    yield
+    app.ssm_provider.clear_cache()  # This will clear SSMProvider cache
+
+
+@pytest.fixture
+def mock_parameter_response(monkeypatch):
+    def mockreturn(name):
+        return "mock_value"
+
+    monkeypatch.setattr(app.ssm_provider, "get", mockreturn)
+
+
+# Pass our fixture as an argument to all tests where we want to mock the get_parameter response
+def test_handler(mock_parameter_response):
+    return_val = app.handler({}, {})
+    assert return_val.get("message") == "mock_value"

docs/examples/utilities/parameters/testing_parameters_fixture.py

@@ -0,0 +1,16 @@
+import pytest
+from src import index
+
+
+@pytest.fixture
+def mock_parameter_response(monkeypatch):
+    def mockreturn(name):
+        return "mock_value"
+
+    monkeypatch.setattr(index.parameters, "get_parameter", mockreturn)
+
+
+# Pass our fixture as an argument to all tests where we want to mock the get_parameter response
+def test_handler(mock_parameter_response):
+    return_val = index.handler({}, {})
+    assert return_val.get("message") == "mock_value"