Реалізація функції зворотного виклику

Нещодавно вам доручили додати зворотні виклики на випадок помилки до DAGʼів, які створює ваша команда. Для початку ви хочете додати простий зворотний виклик при помилці, що записуватиме повідомлення до журналу аудиту, коли sales_etl_dag завершується з помилкою.

Обʼєкти dag і task уже імпортовано, а завдання get_sales_data та process_sales_data створено.

Ця вправа є частиною курсу

Вступ до Apache Airflow на Python

Переглянути курс

Інструкції до вправи

Створіть функцію зворотного виклику з назвою alert_on_failure.
Визначте функцію так, щоб вона приймала будь-які обʼєкти, які Airflow їй передає.
Вкажіть зворотний виклик на випадок помилки, використавши функцію alert_on_failure.

Інтерактивна практична вправа

Спробуйте виконати цю вправу, доповнивши цей зразок коду.

# Create the callback function
def ____(____):
  dag_id = context["dag"].dag_id
  task_id = context["task_instance"].task_id
  print(f"Task {task_id} in Dag {dag_id} has failed.")
  
# Specify the Dag with a failure callback
@dag(dag_id='sales_etl_dag',
     ____=alert_on_failure
)
def sales_etl_dag():
  get_sales_data() >> process_sales_data()
  
sales_etl_dag()

Редагувати та запускати код