使用PythonVirtualenvOperator出现FERNET_KEY配置缺失报错如何解决？

PythonVirtualenvOperator 运行时会启动独立的隔离虚拟环境进程，默认不会继承Airflow主进程的环境变量、配置项，主配置中设置的FERNET_KEY没有被传递到虚拟环境中，导致加密密码解密失败。

• 方法1：单任务显式传递FERNET_KEY
  在DAG定义中直接通过operator的env参数传递密钥，示例代码如下：

  import os
  from airflow.operators.python import PythonVirtualenvOperator
  
  your_task = PythonVirtualenvOperator(
      task_id="your_task_id",
      python_callable=your_callable_func,
      env={"FERNET_KEY": os.environ.get("FERNET_KEY")},
      # 其余原有参数保持不变
      requirements=["your-required-package"],
  )
  

  若你的FERNET_KEY未配置在环境变量中，也可以直接从Airflow配置读取：

  from airflow.configuration import conf
  # 对应env参数改为：
  env={"FERNET_KEY": conf.get("core", "fernet_key")}
  

• 方法2：全局配置虚拟环境自动继承密钥
  打开airflow.cfg配置文件，找到[core]区块，修改virtualenv_env_variables配置项，增加FERNET_KEY相关环境变量即可：

  [core]
  virtualenv_env_variables = AIRFLOW_HOME,AIRFLOW__CORE__FERNET_KEY

  配置修改完成后重启Airflow scheduler和worker服务，后续所有PythonVirtualenvOperator都会自动继承该配置，不需要逐个任务额外设置。

• 方法3：虚拟环境外提前解密（安全性最高）
  如果虚拟环境内的业务逻辑不需要直接调用Airflow的连接、变量读取接口，可以在主进程提前完成解密操作，将明文参数通过op_kwargs传递给虚拟环境内的函数，不需要传递FERNET_KEY到虚拟环境：

  from airflow.hooks.base import BaseHook
  
  # 主进程提前读取解密后的连接信息
  db_conn = BaseHook.get_connection("your_db_conn_id")
  login_user = db_conn.login
  login_pass = db_conn.password
  
  your_task = PythonVirtualenvOperator(
      task_id="your_task_id",
      python_callable=your_callable_func,
      op_kwargs={"login_user": login_user, "login_pass": login_pass},
      # 其余原有参数保持不变
  )
  

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