如何在AWS MWAA的非DAG文件中正常读取自定义OS环境变量

问题根因 #

你遇到的报错本质是MWAA组件的资源加载顺序问题：

• 你通过Airflow插件注入的环境变量，是在插件加载阶段才写入os.environ的
• MWAA的Scheduler解析DAG的时机，早于插件加载完成的时机
• 你的varcheck.py在模块顶层直接读取os.environ["My_Var"]，此时插件还未完成加载，环境变量尚未注入，因此抛出KeyError
• 而VAR = os.environ之所以能正常运行，是因为os.environ是Python进程全局的可变映射对象，你存储的是该对象的引用，插件加载完成后修改该对象的内容，后续访问时自然能读到最新注入的环境变量

注意：请先检查代码中环境变量的键名大小写，报错信息中提到的是MY_VAR（全大写），而你插件中配置的是My_Var（驼峰），避免大小写不匹配导致的额外问题。

解决方案 #

方案1（最优，无需修改公共代码）：使用MWAA原生自定义环境变量 #

MWAA官方支持直接在环境配置中注入OS级环境变量，这类变量会在所有Airflow进程（Scheduler、Worker、Webserver）启动前完成注入，早于DAG解析阶段，完全兼容你现有公共代码的读取逻辑：

1. 进入AWS控制台的MWAA环境管理页面，选择对应环境点击「编辑」
2. 找到「环境变量」配置板块，添加你需要的所有自定义环境变量（包括你当前在插件中配置的PATH、JAVA_HOME、My_Var等）
3. 保存配置并等待环境更新完成即可，无需修改任何业务代码

方案2（兼容插件注入场景，公共代码修改量极小）：懒加载环境变量 #

如果需要保留插件注入环境变量的逻辑，只需修改公共代码中环境变量的读取逻辑，将模块级的立即读取改为延迟加载，不影响其他非Airflow服务的使用：

import os
from typing import Any

class LazyEnv:
    def __init__(self, env_key: str):
        self.key = env_key
        self._val: Any = None

    def __str__(self):
        if self._val is None:
            self._val = os.environ[self.key]
        return self._val

VAR = LazyEnv("My_Var")

该封装对上层调用完全透明，原有调用VAR的逻辑无需做任何修改，同时适配所有运行环境。

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