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

Question

背景

我想限制在我的 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()

但是，如果我想缓存多个 Parameter Store 值，我必须创建多个全局变量并 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 客户端对象。然后将使用缓存的 Boto 客户端调用 get_parameter。例如：

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 客户端是否会导致每个参数仅调用一次 Parameter Store？还是我优化的地方不对？

Answer 1

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

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")

Answer 2

您的原始代码将不起作用，因为 param_util 是一个局部变量，将超出每个 Lambda 调用的范围。

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

装饰器用一个可保存最多 maxsize 最近调用的记忆可调用函数来包装函数。当使用相同的参数定期调用昂贵的或 I/O 绑定的函数时，它可以节省时间。

例子：

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")