Как получить доступ к контексту, такому как execute_date в функции HttpSensor response_check Airflow - PullRequest
Новая
       10

Как получить доступ к контексту, такому как execute_date в функции HttpSensor response_check Airflow

0 голосов
/ 09 мая 2019

Я пишу сообщение о потоке воздуха с помощью датчика Http, который проверяет, включена ли строка даты выполнения в ответ http. Как правильно получить доступ к дате выполнения в функции response_check?

Я пробовал 

s = HttpSensor(
    task_id=some_task_id,
    http_conn_id=some_http_conn_id,
    endpoint=some_endpoint,
    method='POST',
    request_params=some_params_json_str,
    headers={'Content-Type': 'application/json'},
    response_check=lambda response: True if 'dt={{ ds }}' in json.loads(response.text)['list_of_dates'] else False,
    poke_interval=30,
    dag=dag,
)

При тестировании задания оно продолжает тыкать, хотя критерии должны выполняться при первом нажатии. Для сравнения я жестко кодирую response_check по строке даты экспликации даты выполнения, например: 

response_check=lambda response: True if 'dt=2019-05-08' in json.loads(response.text)['list_of_dates'] else False,

и задача вела себя, как и ожидалось, закончилась первым ударом.

Разрешено ли непосредственное использование шаблона '{{ds}}' в функции response_check? Если нет, то как правильно?

Другие вещи, которые я пробовал: 

response_check=lambda response, ds: True if 'dt='+ds in json.loads(response.text)['list_of_dates'] else False,

даст мне 

TypeError: <lambda>() missing 1 required positional argument: 'ds'

Также пробовал: 

response_check=lambda response, **kwargs: True if 'dt='+kwargs['execution_date'] in json.loads(response.text)['result']['partitions'] else False,

даст мне 

KeyError: 'execution_date'
...