Как сделать так, чтобы DAG не работал, когда мы вызываем исключение в нашем коде Python? ⇐ Python

[Anonymous]

Я новичок в группах обеспечения доступности баз данных Airflow. У меня есть скрипт Python для получения вызовов API. Я добавил исключение подъема в свой код: если все вызовы API завершились неудачно, т. е. status_code!=200, тогда должно возникнуть исключение. Однако в журналах воздушного потока это отображается как INFO — [base] Произошла ошибка: все вызовы API завершились неудачно для всех идентификаторов, и DAG завершился успешно. Как сделать так, чтобы группа DAG не удалась, если все вызовы API завершаются неудачно.
Ниже приведен пример кода:

Код: Выделить всё

        response = requests.get(url)
data = response.json()
if response.status_code!=200:
print(f"Error occurred for ID - Response code {response.status_code} {response.reason}")
break
if response.status_code==200:
df1 = pd.concat([df1, pd.json_normalize(data)])
else:
print(f'No data to fetch data for {reaction}:{id}')

if not df1.empty:
all_accounts_failed = False

if all_accounts_failed:
raise Exception("All API calls failed for all IDs")
raise
else:
try:
#rest of the code

Ожидал сбой DAG, однако DAG преуспел, просто отобразив его как INFO.
Фрагмент из DAG

Подробнее здесь: https://stackoverflow.com/questions/793 ... ython-code