技术咨询：lambda_handler函数应嵌入Python主脚本还是独立编写？

不需要单独编写脚本，直接把lambda_handler函数添加到你的main_script.py里即可。这样Lambda可以直接通过这个入口函数触发业务逻辑，结构更简洁，维护成本更低。

调整后的完整代码 #

import logging
import pymysql as pm
import os 
import json

# 配置日志适配Lambda CloudWatch日志系统
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

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=self.get_config()
        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)
                            logger.info("test_function_INSERT_QUERY执行成功，更新%d条记录", cur.rowcount)

                            cur.execute(test_function_EXCEPTIONS_QUERY)
                            logger.info("test_function_EXCEPTIONS_QUERY执行成功，更新%d条异常记录", cur.rowcount)

                        except pm.Error as e:
                            logger.error("数据库错误: %s", e)
                            raise
                        except Exception as e:
                            logger.exception("未知错误")
                            raise
                        else:
                            cnxn.commit()
        test_function(self)

def main():
    # Lambda环境下代码部署在/var/task目录，直接指定配置文件名即可（需确保打包时_config.json与脚本同目录）
    vfc = "_config.json"
    ve=className(vfc)
    ve.run_all()

def lambda_handler(event, context):
    try:
        main()
        return {
            'statusCode': 200,
            'body': json.dumps('任务执行成功')
        }
    except Exception as e:
        logger.error("任务执行失败: %s", e)
        return {
            'statusCode': 500,
            'body': json.dumps(f'任务执行失败: {str(e)}')
        }

if __name__ == "__main__":
    main()

关键注意事项 #

• 配置文件路径：Lambda中代码默认部署到/var/task目录，无需用os.getcwd()拼接路径，直接写配置文件名即可，打包时要确保_config.json与脚本在同一目录。
• 日志规范：把print替换为logging接口，日志会自动同步到CloudWatch，方便后续排查问题。
• 异常处理：在lambda_handler中统一捕获异常，返回明确的状态码，同时抛出异常让Lambda记录完整错误栈。
• 依赖管理：Lambda默认没有pymysql库，需要将依赖与代码一起打包成zip上传，或者使用Lambda层统一管理第三方依赖。

内容的提问来源于stack exchange，提问作者frustrated_gerbil