问题

I don't know how to write a Lambda. Here is my main_script.py that executes 2 stored procedures. It inserts records every day then finds the difference between yesterday's and today's records and writes them to a table.

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

Would I write my lambda_handler function inside my script above or have it as a separate script?

def lambda_handler(event, context):
    #some code

You should have the lambda_handler function in a separate script. The lambda_handler is the entry point for an AWS Lambda function and should be defined in its own Python file. This script will be uploaded and configured as the Lambda function in your AWS environment, and AWS Lambda will invoke this function when the Lambda is triggered.

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

def lambda_handler(event, context):
#some code

答案1

得分: 5

我会把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，花点小钱甚至可以为你轮换密码（因为它与Relational Database Service完全集成）。

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}