Airflow - DAG integrity tests - sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: variable
Airflow - DAG integrity tests - sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: variable
我目前是第一次编写一些气流 DAG 完整性测试。我遇到一个错误，我的一些 operators/tasks 引用了气流变量，例如：
test_var= Variable.get("AIRFLOW_VAR_TEST_VAR")
当我 运行 使用 pytest 进行完整性测试时，出现以下错误：
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) 没有这样的 table: 变量
我可以通过用硬编码值替换 Variable.get 代码来解决这个问题，但想知道是否有更好的方法来处理这个错误？
谢谢，
你应该运行
AIRFLOW__CORE__UNIT_TEST_MODE=True airflow db reset
这将从头开始初始化并重新创建单元测试 sqlite 数据库。
或者，您可以 运行 pytest 使用特定于气流的开关 --with-db-init 开关，它的作用相同。
我目前是第一次编写一些气流 DAG 完整性测试。我遇到一个错误，我的一些 operators/tasks 引用了气流变量，例如：
test_var= Variable.get("AIRFLOW_VAR_TEST_VAR")
当我 运行 使用 pytest 进行完整性测试时，出现以下错误：
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) 没有这样的 table: 变量
我可以通过用硬编码值替换 Variable.get 代码来解决这个问题，但想知道是否有更好的方法来处理这个错误？
谢谢，
你应该运行
AIRFLOW__CORE__UNIT_TEST_MODE=True airflow db reset
这将从头开始初始化并重新创建单元测试 sqlite 数据库。
或者，您可以 运行 pytest 使用特定于气流的开关 --with-db-init 开关，它的作用相同。