我的 lambda 函数是否位于我的主 python 脚本中？

Question

我不知道如何写 Lambda。这是我的 main_script.py，它执行 2 个存储过程。它每天插入记录，然后查找昨天和今天的记录之间的差异并将其写入表中。

import logging
import pymysql as pm
import os 
import json

class className:
    env=None
    config=None

    def __init__(self, env_filename):
        self.env=env_filename
        self.config=self.get_config()

    def get_config(self):
        with open(self.env) as file_in:
            return json.load(file_in)

    def DB_connection(self):
        config=className.get_config(self)
        username=config["exceptions"]["database-secrets"]["aws_secret_username"]
        password=config["exceptions"]["database-secrets"]["aws_secret_password"]
        host=config["exceptions"]["database-secrets"]["aws_secret_host"]
        port=config["exceptions"]["database-secrets"]["aws_secret_port"]
        database=config["exceptions"]["database-secrets"]["aws_secret_db"]

        return pm.connect(
            user=username,
            password=password,
            host=host,
            port=port,
            database=database
        )

    def run_all(self):
        def test_function(self):
            test_function_INSERT_QUERY = "CALL sp_test_insert();"
            test_function_EXCEPTIONS_QUERY = "CALL sp_test_exceptions();"
            test = self.config["exceptions"]["functions"]["test_function"]
            if test:
                with self.DB_connection() as cnxn:
                    with cnxn.cursor() as cur:
                        try:
                            cur.execute(test_function_INSERT_QUERY)
                            print("test_function_INSERT_QUERY insertion query ran successfully, {} records updated.".format(cur.rowcount))

                            cur.execute(test_function_EXCEPTIONS_QUERY)
                            print("test_function_EXCEPTIONS_QUERY exceptions query ran successfully, {} exceptions updated.".format(cur.rowcount))

                        except pm.Error as e:
                            print(f"Error: {e}")

                        except Exception as e:
                            logging.exception(e)

                        else:
                            cnxn.commit()
        test_function(self)

def main():
    cwd=os.getcwd()
    vfc=(cwd+"\_config"+".json")
    ve=className(vfc)
    ve.run_all()
if __name__ == "__main__":
    main()

我会在上面的脚本中编写 lambda_handler 函数还是将其作为单独的脚本？

def lambda_handler(event, context):
    #some code

Answer 1

我将其视为lambda_handler(event, context)等同于，main() 但您不需要if __name__ ...子句，因为您从不从控制台运行 lambda 函数。

您还需要使用boto3库来抽象 AWS 服务及其功能。查看教程以开始使用。

作为首要任务，我会将数据库凭据从文件系统中取出并放入安全的数据存储中。您当然可以配置 Lambda 环境变量，但Systems Manager Parameter Store更安全并且非常容易从代码中调用，例如：

import boto3
ssm = boto3.client('ssm', region_name='us-east-1')

def lambda_handler(event, context):
    password = ssm.get_parameters(Names=['/pathto/password'], WithDecryption=True)['Parameters'][0]['Value']
    return {"password": password}

还有一个更高级的选项，Secrets Manager，只需花一点钱，它甚至可以为您轮换密码（因为它与关系数据库服务完全集成）。