Airflow 2[Cloud Composer]自定义插件导入本地与线上兼容问题

两种环境的Python模块搜索路径存在差异：

• Cloud Composer 环境会自动将GCS桶内的plugins目录添加到Python的sys.path搜索路径中，因此可以直接识别cse根包，不需要带plugins前缀。
• 本地测试环境默认只会将执行测试命令的当前目录加入搜索路径，无法直接搜索到plugins目录下的cse包，因此需要带plugins前缀才能正常导入。

方案1：统一配置本地模块搜索路径（最推荐） #

无需修改业务代码的导入语句，仅调整本地测试的路径配置即可：
如果你使用pytest框架，直接在项目根目录新建pytest.ini配置文件，添加如下内容：

[pytest]
pythonpath = plugins

如果是其他测试框架或者需要手动配置，在测试的入口文件开头添加路径注入代码：

import sys
from pathlib import Path

# 调整路径对应你本地项目的根目录位置
sys.path.insert(0, str(Path(__file__).parent.parent / "plugins"))

配置完成后，统一使用from cse.operators.cse_to_bq import CSEToBQOperator的导入方式，本地测试和Cloud Composer环境都可以正常运行。

方案2：导入兼容 fallback #

如果不想修改测试配置，可以直接在DAG文件中使用异常捕获的兼容导入逻辑：

try:
    # 适配 Cloud Composer 环境
    from cse.operators.cse_to_bq import CSEToBQOperator
except ImportError:
    # 适配本地测试环境
    from plugins.cse.operators.cse_to_bq import CSEToBQOperator

这种方式无需额外配置，代码修改量小，也能同时兼容两种环境。

你提供的目录结构中plugins/cse/operators.py下方还有cse_to_bq.py，这里的operators应该是目录而非.py文件，如果实际是文件的话需要调整为目录结构，否则会导致导入失败。

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