Airflow自定义视图执行airflow list_dags触发subprocess报错求助

我在Apache Airflow 1.10.12中创建了一个自定义视图，打开页面时Web端抛出如下错误：

Traceback (most recent call last):
  File "/home/deploy/airflow_env/lib/python3.7/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/home/deploy/airflow_env/lib/python3.7/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/home/deploy/airflow_env/lib/python3.7/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/home/deploy/airflow_env/lib/python3.7/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/home/deploy/airflow_env/lib/python3.7/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/home/deploy/airflow_env/lib/python3.7/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/home/deploy/airflow_env/lib/python3.7/site-packages/flask_admin/base.py", line 69, in inner
    return self._run_view(f, *args, **kwargs)
  File "/home/deploy/airflow_env/lib/python3.7/site-packages/flask_admin/base.py", line 368, in _run_view
    return fn(self, *args, **kwargs)
  File "/home/deploy/airflow/plugins/views/backfill_view.py", line 14, in base
    dags=self._get_dag_names()
  File "/home/deploy/airflow/plugins/views/backfill_view.py", line 66, in _get_dag_names
    'airflow list_dags', shell=True
  File "/usr/lib/python3.7/subprocess.py", line 411, in check_output
    **kwargs).stdout
  File "/usr/lib/python3.7/subprocess.py", line 512, in run
    output=stdout, stderr=stderr)
subprocess.CalledProcessError: Command 'airflow list_dags' returned non-zero exit status 120.

对应的代码片段（backfill_view.py第66行）：

dags_raw = subprocess.check_output(
    'airflow list_dags', shell=True
).split()

我可以在服务器上直接运行airflow list_dags命令且执行正常，但通过自定义视图调用时出现上述错误，恳请提供帮助。

环境信息：

Apache Airflow [1.10.12]
Platform: [Linux, x86_64]
Python Version: [3.7.3]

解决方案 #

1. 使用完整的airflow命令路径 #

手动执行which airflow获取命令的绝对路径，替换代码中的airflow为完整路径，确保subprocess能找到正确的命令：

dags_raw = subprocess.check_output(
    '/home/deploy/airflow_env/bin/airflow list_dags', shell=True
).split()

2. 显式传递正确的环境变量 #

Airflow Webserver的环境变量可能和你登录shell后的环境不一致，需要手动传递必要的环境变量（如AIRFLOW_HOME、PATH）：

import os

# 复制当前环境变量并补充必要配置
env = os.environ.copy()
env['AIRFLOW_HOME'] = '/path/to/your/airflow/home'  # 替换为你的Airflow主目录
env['PATH'] = f"/home/deploy/airflow_env/bin:{env['PATH']}"  # 确保虚拟环境的bin目录在PATH最前面

dags_raw = subprocess.check_output(
    'airflow list_dags', shell=True, env=env
).split()

3. 改用Airflow内部API获取DAG列表（推荐） #

直接调用Airflow的内部模型获取DAG列表，避免依赖shell命令和环境变量问题：

from airflow.models import DagBag

def _get_dag_names(self):
    dag_bag = DagBag()
    # 返回所有可用的DAG ID列表
    return list(dag_bag.dags.keys())

4. 检查Webserver的运行权限和用户 #

确保Airflow Webserver进程的运行用户与你手动执行命令的用户一致，且该用户有访问Airflow配置文件、DAG目录的权限。

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