如何在 AWS Lambda 中缓存多个 AWS Parameter Store 值?

How do I cache multiple AWS Parameter Store values in an AWS Lambda?

背景

我想限制在我的 AWS Lambda 中调用 AWS Parameter Store 的次数。使用全局变量,我在第一次调用 Parameter Store 时缓存了一个 Parameter Store 值。

main.py

import os

import boto3


redis_password = None

def get_redis_password():
    global redis_password
    if not redis_password:
        client = boto3.client("ssm")
        redis_password = client.get_parameter(
            Name=f"{os.environ["ENV"]}.redis-cache.password",
            WithDecryption=True
        )
    return redis_password["Parameter"]["Value"]

def lambda_handler(event, context):
    get_redis_password()

但是,如果我想缓存多个参数存储值,我必须创建多个全局变量和 if not [INSERT_GLOBAL_VARIABLE] 检查。例如:

main.py

import os

import boto3


redis_password = None
another_parameter_store_value = None

def get_redis_password():
    global redis_password
    if not redis_password:
        client = boto3.client("ssm")
        redis_password = client.get_parameter(
            Name=f"{os.environ["ENV"]}.redis-cache.password",
            WithDecryption=True
        )
    return redis_password["Parameter"]["Value"]

def get_another_parameter_store_value():
    global another_parameter_store_value
    if not another_parameter_store_value:
        client = boto3.client("ssm")
        another_parameter_store_value = client.get_parameter(
            Name=f"{os.environ["ENV"]}.another.parameter.store.key",
            WithDecryption=True
        )
    return redis_password["Parameter"]["Value"]

def lambda_handler(event, context):
    get_redis_password()
    get_another_parameter_store_value()

尝试的解决方案

为了解决这个问题,我创建了一个 Parameter Store 实用程序。

parameter_util.py

import os
import boto3


class ParameterUtil:
    def __init__(self):
        self.boto_client = boto3.client("ssm")

    def get_parameter(self, parameter_path):
        response = self.boto_client.get_parameter(
            Name=f"{os.environ['ENV']}.{parameter_path}", WithDecryption=True
        )

        return response["Parameter"]["Value"]

我的理论是,通过将 AWS Boto 客户端实例化为实例变量,它将缓存整个 Boto 客户端对象。然后 get_parameter 将使用缓存的 Boto 客户端调用。例如:

main.py

import os

import boto3

from parameter_util import ParameterUtil


redis_password = None

def get_redis_password():
    global redis_password
    if not redis_password:
        client = boto3.client("ssm")
        redis_password = client.get_parameter(
            Name=f"{os.environ["ENV"]}.redis-cache.password",
            WithDecryption=True
        )
    return redis_password["Parameter"]["Value"]

def lambda_handler(event, context):
    param_util = ParameterUtil()
    param_util.get_parameter(".redis-cache.password")
    param_util.get_parameter(".another.parameter.store.key")

但是,我不确定这是否能解决问题。

问题

在调用 get_parameter 时,缓存 Boto 客户端是否会导致每个参数仅调用一次参数存储?还是我优化错了地方?

我喜欢这种方法。我可能会建议将其抽象为如下内容:

main.py

parameter_store_values = {}
client = boto3.client("ssm")

def lookup_function(key):
    global parameter_store_values
    global client
    if parameter_store_values.get(key) is None:
        value = client.get_parameter(
            Name=key,
            WithDecryption=True)["Parameter"]["Value"]
        parameter_store_values[key] = value
    return value

def lambda_handler(event, context):
    redis_password = lookup_function(f"{os.environ["ENV"]}.redis-cache.password")
    another_parameter_store_key = lookup_function(f"{os.environ["ENV"]}.another.parameter.store.key")

您的原始代码将无法运行,因为 param_util 是一个局部变量,它将超出每次 Lambda 调用的范围。

您可以使用内置的 @functools.lru_cache 创建一个处理任何参数的简单函数。它将根据函数的输入为您缓存 return 值 (Python 3.2+).

Decorator to wrap a function with a memoizing callable that saves up to the maxsize most recent calls. It can save time when an expensive or I/O bound function is periodically called with the same arguments.

示例:

ssm_client = boto3.client("ssm")

@lru_cache(maxsize=None)
def get_param(name):
    return ssm_client.get_parameter(
        Name=f"{os.environ['ENV']}.{name}",
        WithDecryption=True
    )["Parameter"]["Value"]

def lambda_handler(event, context):
  redis_password = get_param("redis-cache.password")
  another_parameter_store_key = get_param("another.parameter.store.key")