fff_checker_v1.py

import pendulum
from airflow import DAG
from airflow.decorators import dag, task
from airflow.operators.empty import EmptyOperator
from airflow.providers.http.operators.http import SimpleHttpOperator
from airflow.providers.telegram.operators.telegram import TelegramOperator
from airflow.utils.edgemodifier import Label
from airflow.utils.trigger_rule import TriggerRule
from airflow.utils.session import create_session
from airflow.models import Connection


HTTP_CONN_ID = 'factorio_friday_facts'


@dag(schedule_interval='0 0 * * 6', start_date=pendulum.datetime(2020, 11, 1), catchup=False)
def fff_checker_v1():
    start = EmptyOperator(task_id='start')
    end = EmptyOperator(task_id='end', trigger_rule=TriggerRule.ONE_SUCCESS)

    yes = EmptyOperator(task_id='yes_news')
    no = EmptyOperator(task_id='no_news', trigger_rule=TriggerRule.ALL_FAILED)

    check_post = SimpleHttpOperator(
        task_id='check_post',
        http_conn_id=HTTP_CONN_ID,
        endpoint='',
        method='GET',
    )

    telegram = TelegramOperator(
        task_id='send_message_telegram',
        telegram_conn_id='tg_matvey_to_me',
        text=f'Хуя себе там похоже новости factorio пришли http://{{{{ conn.{HTTP_CONN_ID}.host }}}}!',
    )

    @task
    def update_connection():
        with create_session() as session:
            conn = session.query(Connection).filter(Connection.conn_id == HTTP_CONN_ID).first()
            print(f'Mingling with connection {conn} (updating host from {conn.host})')
            conn.host = f'{conn.host[:-3]}{int(conn.host[-3:]) + 1}'
            print(f'New host: {conn.host}')
            session.add(conn)
            session.commit()


    start >> check_post >> Label('404 not found') >> no >> end
    check_post >> Label('Yay update') >> yes >> telegram >> update_connection() >> end


actual_dag = fff_checker_v1()