Create dag for soroban pipelines

airflow_variables_dev.json

@@ -91,10 +91,23 @@
   "dbt_client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/803774498738-compute%40developer.gserviceaccount.com",
   "dbt_dataset": "test_sdf",
   "dbt_full_refresh_models": {
-    "partnership_assets__account_holders_activity_fact": true,
-    "partnership_assets__asset_activity_fact": true
+    "partnership_assets__account_holders_activity_fact": false,
+    "partnership_assets__asset_activity_fact": false,
+    "int_soroban__contract_metrics_day": false,
+    "int_soroban__contract_metrics_week": false,
+    "int_soroban__contract_metrics_month": false,
+    "int_soroban__contract_metrics_year": false,
+    "int_soroban__top_contract_activity_day": false,
+    "int_soroban__top_contract_activity_week": false,
+    "int_soroban__top_contract_activity_month": false,
+    "int_soroban__top_contract_activity_year": false,
+    "contract_assets_fact": true,
+    "contract_type_fact": true,
+    "expiration_current": true,
+    "soroban__contract_metrics_agg": false,
+    "soroban__top_contract_activity_agg": false
   },
-  "dbt_image_name": "stellar/stellar-dbt:2ef9be7",
+  "dbt_image_name": "chowbao/stellar-dbt:soroban2",
   "dbt_job_execution_timeout_seconds": 300,
   "dbt_job_retries": 1,
   "dbt_keyfile_profile": "",
@@ -103,7 +116,7 @@
   "dbt_private_key": "dummy",
   "dbt_private_key_id": "dummy",
   "dbt_project": "test-hubble-319619",
-  "dbt_target": "dev_airflow",
+  "dbt_target": "test",
   "dbt_threads": 1,
   "dbt_token_uri": "https://oauth2.googleapis.com/token",
   "gcs_exported_data_bucket_name": "us-central1-hubble-1pt5-dev-7db0e004-bucket",

dags/soroban_tables_dag.py

@@ -0,0 +1,116 @@
+import datetime
+
+from airflow import DAG
+from airflow.models.variable import Variable
+from stellar_etl_airflow.build_cross_dependency_task import build_cross_deps
+from stellar_etl_airflow.build_dbt_task import build_dbt_task
+from stellar_etl_airflow.default import get_default_dag_args, init_sentry
+
+init_sentry()
+
+dag = DAG(
+    "soroban_tables",
+    default_args=get_default_dag_args(),
+    start_date=datetime.datetime(2023, 9, 30),
+    description="This DAG runs dbt to create the tables for the models in marts/soroban.",
+    schedule_interval="0 17 * * *",  # Daily 11 AM UTC
+    params={},
+    catchup=True,
+    max_active_runs=1,
+)
+
+# tasks for staging tables
+stg_config_settings = build_dbt_task(dag, "stg_config_settings")
+stg_contract_code = build_dbt_task(dag, "stg_contract_code")
+stg_contract_data = build_dbt_task(dag, "stg_contract_data")
+stg_expiration = build_dbt_task(dag, "stg_expiration")
+
+# tasks for intermediate soroban aggregate tables
+int_soroban__enriched_history_operations_year = build_dbt_task(
+    dag, "int_soroban__enriched_history_operations_year"
+)
+
+int_soroban__contract_metrics_day = build_dbt_task(
+    dag, "int_soroban__contract_metrics_day"
+)
+int_soroban__contract_metrics_day = build_dbt_task(
+    dag, "int_soroban__contract_metrics_week"
+)
+int_soroban__contract_metrics_day = build_dbt_task(
+    dag, "int_soroban__contract_metrics_month"
+)
+int_soroban__contract_metrics_day = build_dbt_task(
+    dag, "int_soroban__contract_metrics_year"
+)
+
+int_soroban__top_contract_activity_day = build_dbt_task(
+    dag, "int_soroban__top_contract_activity_day"
+)
+int_soroban__top_contract_activity_week = build_dbt_task(
+    dag, "int_soroban__top_contract_activity_week"
+)
+int_soroban__top_contract_activity_month = build_dbt_task(
+    dag, "int_soroban__top_contract_activity_month"
+)
+int_soroban__top_contract_activity_year = build_dbt_task(
+    dag, "int_soroban__top_contract_activity_year"
+)
+
+# tasks for marts tables
+contract_assets_fact = build_dbt_task(dag, "contract_assets_fact")
+contract_type_fact = build_dbt_task(dag, "contract_type_fact")
+expiration_current = build_dbt_task(dag, "expiration_current")
+soroban__contract_metrics_agg = build_dbt_task(dag, "soroban__contract_metrics_agg")
+soroban__top_contract_activity_agg = build_dbt_task(
+    dag, "soroban__top_contract_activity_agg"
+)
+
+# DAG task graph
+stg_config_settings
+stg_contract_code
+
+stg_contract_data >> contract_assets_fact
+contract_type_fact
+stg_expiration >> expiration_current
+
+(
+    int_soroban__enriched_history_operations_year
+    >> int_soroban__contract_metrics_day
+    >> soroban__contract_metrics_agg
+)
+(
+    int_soroban__enriched_history_operations_year
+    >> int_soroban__contract_metrics_day
+    >> soroban__contract_metrics_agg
+)
+(
+    int_soroban__enriched_history_operations_year
+    >> int_soroban__contract_metrics_day
+    >> soroban__contract_metrics_agg
+)
+(
+    int_soroban__enriched_history_operations_year
+    >> int_soroban__contract_metrics_day
+    >> soroban__contract_metrics_agg
+)
+
+(
+    int_soroban__enriched_history_operations_year
+    >> int_soroban__top_contract_activity_day
+    >> soroban__top_contract_activity_agg
+)
+(
+    int_soroban__enriched_history_operations_year
+    >> int_soroban__top_contract_activity_week
+    >> soroban__top_contract_activity_agg
+)
+(
+    int_soroban__enriched_history_operations_year
+    >> int_soroban__top_contract_activity_month
+    >> soroban__top_contract_activity_agg
+)
+(
+    int_soroban__enriched_history_operations_year
+    >> int_soroban__top_contract_activity_year
+    >> soroban__top_contract_activity_agg
+)