feat: add get_parameter utility

Author: nmoutschen

aws_lambda_powertools/utilities/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+
+"""General utilities for Powertools"""
+
+
+from .parameters import get_parameter
+
+__all__ = ["get_parameter"]

aws_lambda_powertools/utilities/parameters.py

@@ -0,0 +1,63 @@
+"""
+Parameter retrieval and caching utility
+"""
+
+
+from collections import namedtuple
+from datetime import datetime, timedelta
+
+import boto3
+
+DEFAULT_MAX_AGE = 5
+ExpirableValue = namedtuple("ExpirableValue", ["value", "ttl"])
+PARAMETER_VALUES = {}
+ssm = boto3.client("ssm")
+
+
+def get_parameter(name: str, max_age: int = DEFAULT_MAX_AGE) -> str:
+    """
+    Retrieve a parameter from the AWS Systems Manager (SSM) Parameter Store
+
+    This will keep a local version in cache for `max_age` seconds to prevent
+    overfetching from SSM Parameter Store.
+
+    See the [AWS Systems Manager Parameter Store documentation]
+    (https://docs.aws.amazon.com/systems-manager/latest/userguide/systems-manager-parameter-store.html)
+    for more information.
+
+    Parameters
+    ----------
+    name: str
+        Name of the SSM Parameter
+    max_age: int
+        Duration for which the parameter value can be cached
+
+    Example
+    -------
+
+        from aws_lambda_powertools.utilities import get_parameter
+
+        def lambda_handler(event, context):
+            # This will only make a call to the SSM service every 30 seconds.
+            value = get_parameter("my-parameter", max_age=30)
+
+    Raises
+    ------
+    ssm.exceptions.InternalServerError
+        When there is an internal server error from AWS Systems Manager
+    ssm.exceptions.InvalidKeyId
+        When the key ID is invalid
+    ssm.exceptions.ParameterNotFound
+        When the parameter name is not found in AWS Systems Manager
+    ssm.exceptions.ParameterVersionNotFound
+        When a version of the parameter is not found in AWS Systems Manager
+    """
+
+    if name not in PARAMETER_VALUES or PARAMETER_VALUES[name].ttl < datetime.now():
+        # Retrieve the parameter from AWS Systems Manager
+        parameter = ssm.get_parameter(Name=name)
+        PARAMETER_VALUES[name] = ExpirableValue(
+            parameter["Parameter"]["Value"], datetime.now() + timedelta(seconds=max_age)
+        )
+
+    return PARAMETER_VALUES[name].value

poetry.lock

@@ -21,6 +21,7 @@ license = "MIT-0"
 python = "^3.6"
 aws-xray-sdk = "^2.5.0"
 fastjsonschema = "~=2.14.4"
+boto3 = "^1.12"
 
 [tool.poetry.dev-dependencies]
 coverage = {extras = ["toml"], version = "^5.0.3"}

pyproject.toml

@@ -0,0 +1,128 @@
+import random
+import string
+from datetime import datetime, timedelta
+
+import pytest
+from botocore import stub
+
+from aws_lambda_powertools import utilities
+from aws_lambda_powertools.utilities import parameters
+
+
+@pytest.fixture(scope="function")
+def mock_name():
+    # Parameter name must match [a-zA-Z0-9_.-/]+
+    return "".join(random.choices(string.ascii_letters + string.digits + "_.-/", k=random.randrange(3, 200)))
+
+
+@pytest.fixture(scope="function")
+def mock_value():
+    # Standard parameters can be up to 4 KB
+    return "".join(random.choices(string.printable, k=random.randrange(100, 4000)))
+
+
+@pytest.fixture(scope="function")
+def mock_version():
+    return random.randrange(1, 1000)
+
+
+def test_get_parameter_new(monkeypatch, mock_name, mock_value, mock_version):
+    """
+    Test get_parameter() with a new parameter name
+    """
+
+    # Patch the parameter value store
+    monkeypatch.setattr(parameters, "PARAMETER_VALUES", {})
+
+    # Stub boto3
+    stubber = stub.Stubber(parameters.ssm)
+    response = {
+        "Parameter": {
+            "Name": mock_name,
+            "Type": "String",
+            "Value": mock_value,
+            "Version": mock_version,
+            "Selector": f"{mock_name}:{mock_version}",
+            "SourceResult": "string",
+            "LastModifiedDate": datetime(2015, 1, 1),
+            "ARN": f"arn:aws:ssm:us-east-2:111122223333:parameter/{mock_name}",
+        }
+    }
+    expected_params = {"Name": mock_name}
+    stubber.add_response("get_parameter", response, expected_params)
+    stubber.activate()
+
+    # Get the parameter value
+    try:
+        value = utilities.get_parameter(mock_name)
+
+        assert value == mock_value
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
+def test_get_parameter_cached(monkeypatch, mock_name, mock_value, mock_version):
+    """
+    Test get_parameter() with a cached value for parameter name
+    """
+
+    # Patch the parameter value store
+    monkeypatch.setattr(
+        parameters,
+        "PARAMETER_VALUES",
+        {mock_name: parameters.ExpirableValue(mock_value, datetime.now() + timedelta(seconds=60))},
+    )
+
+    # Stub boto3
+    stubber = stub.Stubber(parameters.ssm)
+    stubber.activate()
+
+    # Get the parameter value
+    try:
+        value = utilities.get_parameter(mock_name)
+
+        assert value == mock_value
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()
+
+
+def test_get_parameter_expired(monkeypatch, mock_name, mock_value, mock_version):
+    """
+    Test get_parameter() with a cached, but expired value for parameter name
+    """
+
+    # Patch the parameter value store
+    monkeypatch.setattr(
+        parameters,
+        "PARAMETER_VALUES",
+        {mock_name: parameters.ExpirableValue(mock_value, datetime.now() - timedelta(seconds=60))},
+    )
+
+    # Stub boto3
+    stubber = stub.Stubber(parameters.ssm)
+    response = {
+        "Parameter": {
+            "Name": mock_name,
+            "Type": "String",
+            "Value": mock_value,
+            "Version": mock_version,
+            "Selector": f"{mock_name}:{mock_version}",
+            "SourceResult": "string",
+            "LastModifiedDate": datetime(2015, 1, 1),
+            "ARN": f"arn:aws:ssm:us-east-2:111122223333:parameter/{mock_name}",
+        }
+    }
+    expected_params = {"Name": mock_name}
+    stubber.add_response("get_parameter", response, expected_params)
+    stubber.activate()
+
+    # Get the parameter value
+    try:
+        value = utilities.get_parameter(mock_name)
+
+        assert value == mock_value
+        stubber.assert_no_pending_responses()
+    finally:
+        stubber.deactivate()