在调用气流测试中设置 dag_run.conf 参数

Question

有谁知道在bash提示符下运行airflow test时是否有办法设置dag_run.conf参数？

例如，我已经从官方气流存储库下载了example_trigger_target_dag，我想测试run_this任务。通常我会执行以下操作：

~/$ airflow test example_trigger_target_dag run_this '2018-01-01'

但是，运行此操作会产生错误：

--------------------------------------------------------------------------------
Starting attempt 1 of 1
--------------------------------------------------------------------------------
[2018-05-02 10:50:01,154] {models.py:1342} INFO - Executing <Task(PythonOperator): run_this> on 2018-01-01 00:00:00
[2018-05-02 10:50:01,262] {models.py:1417} ERROR - 'NoneType' object has no attribute 'conf'
Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line 1374, in run
result = task_copy.execute(context=context)
File "/usr/local/lib/python2.7/dist-packages/airflow/operators/python_operator.py", line 80, in execute
return_value = self.python_callable(*self.op_args, **self.op_kwargs)
File "/home/annalect/uk_ds_airflow/dags/playpen/example_trigger_target_dag.py", line 56, in run_this_func
print("Remotely received value of {} for key=message".format(kwargs['dag_run'].conf['message']))
AttributeError: 'NoneType' object has no attribute 'conf'

我已经尝试使用task_params参数，但是我要么语法错误，要么它没有达到我想要的效果，因为它会产生与上述相同的错误：

~/$ airflow test --task_params '{"kwargs": {"dag_run": {"conf": {"message": "Hey world"}}}}' example_trigger_target_dag run_this '2018-01-01'

[2018-05-02 11:10:58,065] {models.py:1441} INFO - Marking task as FAILED.
[2018-05-02 11:10:58,070] {models.py:1462} ERROR - 'NoneType' object has no attribute 'conf'

那么有谁知道如何测试依赖于dag_run.conf值的任务呢？

谢谢！

Answer 1

airflow test命令没有--conf选项，但您可以通过将参数传递给任务的python_callable来解决此问题。

在可调用对象中，如果设置了kwargs['test_mode']，则可以检索参数以构建虚拟DagRun对象，如下所示：

from airflow.models import DagRun
...
def run_this_func(ds, **kwargs):
if kwargs['test_mode']:
kwargs['dag_run'] = DagRun(conf=kwargs['params'])
print("Remotely received value of {} for key=message".format(kwargs['dag_run'].conf['message']))

要测试example_trigger_target_dag，只需执行以下操作：

airflow test example_trigger_target_dag test_trigger_dagrun "2018-01-01" -tp '{"message":"Hello world"}'

您将获得：

Remotely received value of Hello world for key=message

现在，您可以编写装饰器，而不是将测试代码放在任务中。另外，由于我们只是使用DagRun的conf属性，我们不妨使用SimpleNamespace。 最后，为了避免在查找kwargs时出现潜在的键错误，我们可以将get与默认值一起使用。

from types import SimpleNamespace
def allow_conf_testing(func):
def wrapper(*args, **kwargs):
if kwargs.get('test_mode', False):
kwargs['dag_run'] = SimpleNamespace(conf=kwargs.get('params', {}))
func(*args, **kwargs)
return wrapper
@allow_conf_testing
def run_this_func(ds, **kwargs):
print("Remotely received value of {} for key=message".format(kwargs['dag_run'].conf['message']))