website_sync.py

import sys
import json
from os import path
from datetime import datetime, timedelta
import logging

from airflow import DAG
from airflow.decorators import dag, task
from airflow.operators.python import get_current_context
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago
from airflow.providers.http.operators.http import SimpleHttpOperator
# HACK: Fix for loading relative modules.
sys.path.append(path.dirname(path.realpath(__file__)))
from tasks.airbyte import fetch_airbyte_connections_tg
from providers.airbyte.operator import AirbyteTriggerSyncOperator
from tasks.alerting import send_alert_discord
from tasks.config import INTERNAL_ENV

"""
DAG to sync data from github to the Website API
"""

logging.basicConfig(stream=sys.stdout, level=logging.INFO)

ARGS = {
    'owner': 'apentori',
    'depends_on_past': False,
    'start_date': datetime(2023,6,1),
    'email': ['alexis@status.im'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': timedelta(minutes=10),
    'on_failure_callback': send_alert_discord,
}

airbyte_connections=['gh_sync_website_repos', 'load_hasura_website']

@dag('github_website_sync',
            default_args=ARGS,
            # Run every 30 minutes
            schedule_interval='0 */6 * * *')
def github_website_sync():
    connections_id=fetch_airbyte_connections_tg(airbyte_connections)

    # Trigger Airbyte fetch Data from Github
    gh_sync_website_repos = AirbyteTriggerSyncOperator(
        task_id='airbyte_fetch_github',
        airbyte_conn_id='airbyte_conn',
        connection_id=connections_id['gh_sync_website_repos'],
        asynchronous=False,
        wait_seconds=3
    )

    # Launch DBT transformation on the data previously fetched
    dbt_transform = BashOperator(
        task_id='dbt_run_models_github',
        bash_command='dbt run --select github',
        env=INTERNAL_ENV,
        append_env=True
    )
    # Trigger Airbyte Sync from main database to Hasura
    load_hasura = AirbyteTriggerSyncOperator(
        task_id='airbyte_sync_hasura',
        airbyte_conn_id='airbyte_conn',
        connection_id=connections_id['load_hasura_website'],
        asynchronous=False,
        wait_seconds=3
    )

    connections_id >> gh_sync_website_repos >> dbt_transform >> load_hasura

github_website_sync()