marketing-metabase-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

"""
DAG to extract data for the marketing dashboard in Metabase
"""

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

ARGS = {
    'owner': 'apentori',
    'depends_on_past': False,
    'start_date': datetime(2025,2,5),
    '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=[
    # For prod, need to add Fathom sync
    'marketing-fathom-extract',
    # Syncing with metabase
    'sync_metabase_fathom',
    'sync_metabase_twitter',
    'sync_metabase_simplecast'
]

@dag('marketing_metabase_export',
            default_args=ARGS,
            schedule_interval='0 6 * * *')
def marketing_metabase_export():
    connections_id=fetch_airbyte_connections_tg(airbyte_connections)

    fathom_export = AirbyteTriggerSyncOperator(
        task_id='airbyte_fathom_extract',
        airbyte_conn_id='airbyte_conn',
        connection_id=connections_id['marketing-fathom-extract'],
        asynchronous=False,
        wait_seconds=3
    )
    fathom_sync = AirbyteTriggerSyncOperator(
        task_id='airbyte_fathom_sync',
        airbyte_conn_id='airbyte_conn',
        connection_id=connections_id['sync_metabase_fathom'],
        asynchronous=False,
        wait_seconds=3
    )

    twitter_sync = AirbyteTriggerSyncOperator(
        task_id='airbyte_twitter_sync',
        airbyte_conn_id='airbyte_conn',
        connection_id=connections_id['sync_metabase_twitter'],
        asynchronous=False,
        wait_seconds=3
    )

    simplecast_sync = AirbyteTriggerSyncOperator(
        task_id='airbyte_simplecast_sync',
        airbyte_conn_id='airbyte_conn',
        connection_id=connections_id['sync_metabase_simplecast'],
        asynchronous=False,
        wait_seconds=3
    )
    connections_id >> fathom_export >> fathom_sync >> twitter_sync >> simplecast_sync

marketing_metabase_export()