跨Python文件调用函数时异常捕获失败问题求助

一、为什么watcher.py无法捕获异常？ #

你当前的代码中，watcher.py的except块捕获到异常后立刻重新抛出（raise e），导致异常最终还是会向上传递到顶层，看起来像是没被捕获。要实现真正的捕获，需要在except块中处理异常（比如打印日志、执行容错逻辑），而非重新抛出。

二、修正异常捕获的代码示例 #

修改后的watcher.py #

import main_etl

def main_function():
    try:
        main_etl.main(arg1='', arg2='', arg3='')
    except Exception as e:
        # 此处处理异常，而非重新抛出
        print(f"捕获到异常: {str(e)}")
        # 可添加重试、日志记录等容错逻辑
    else:
        print("main函数执行成功，执行后续操作...")

# 务必调用主函数
if __name__ == "__main__":
    main_function()

补充：简化main_etl.py的冗余异常处理 #

你的main_etl.py中，test_function和main的try-except仅捕获后重新抛出，无额外处理，属于冗余代码，可直接删除，让异常自然向上传递，不影响watcher的捕获：

def test_function():
    # 直接执行业务逻辑，异常自动向上传递
    do something....

def main(arg1='', arg2='', arg3=''):
    test_function()

三、正确调用main()函数的两种方式 #

方式1：模块导入调用（推荐，进程内调用） #

即你当前使用的方式，直接导入main_etl模块后调用main()函数。优势：

• 进程内调用，性能更高
• 可直接传递Python对象作为参数
• 异常可跨函数/模块传递，便于捕获和处理

注意：若main_etl.py中有if __name__ == "__main__"的执行逻辑，导入时不会触发，仅直接运行main_etl.py时才会执行。

方式2：子进程调用（os.system/subprocess，进程隔离） #

若需要完全的进程隔离（避免main_etl的代码影响watcher运行环境），可使用subprocess模块调用：

第一步：修改main_etl.py，支持命令行参数

让main()函数可接收命令行参数：

import sys

def test_function():
    do something....

def main(arg1='', arg2='', arg3=''):
    test_function()

if __name__ == "__main__":
    # 从命令行参数获取arg1、arg2、arg3，默认为空字符串
    arg1 = sys.argv[1] if len(sys.argv) >=2 else ''
    arg2 = sys.argv[2] if len(sys.argv) >=3 else ''
    arg3 = sys.argv[3] if len(sys.argv) >=4 else ''
    main(arg1, arg2, arg3)

第二步：在watcher.py中用subprocess调用

import subprocess

def main_function():
    try:
        # 调用main_etl.py并传递参数
        result = subprocess.run(
            ['python', 'main_etl.py', 'value1', 'value2', 'value3'],
            capture_output=True,
            text=True,
            check=True  # 进程返回非0码时，抛出CalledProcessError
        )
        print("执行成功，输出:", result.stdout)
    except subprocess.CalledProcessError as e:
        print(f"子进程执行失败，返回码: {e.returncode}，错误输出: {e.stderr}")
    except Exception as e:
        print(f"其他异常: {str(e)}")
    else:
        print("执行后续操作...")

if __name__ == "__main__":
    main_function()

说明：

• os.system()仅能获取返回码，无法捕获输出，推荐使用更灵活的subprocess.run()
• 子进程的异常不会直接传递到父进程，需通过返回码和stderr判断执行状态

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