Update Canvas sandbox dependencies

airflow_variables_dev.json

@@ -55,7 +55,7 @@
   "dbt_image_name": "stellar/stellar-dbt:6ee7342",
   "dbt_job_execution_timeout_seconds": 300,
   "dbt_job_retries": 1,
-  "dbt_mart_dataset": "test_sdf_marts",
+  "dbt_mart_dataset": "test_crypto_stellar_dbt",
   "dbt_maximum_bytes_billed": 250000000000,
   "dbt_project": "test-hubble-319619",
   "dbt_target": "test",
@@ -251,9 +251,13 @@
   "dbt_tables": {
     "signers_current": "account_signers_current",
     "accounts_current": "accounts_current",
+    "config_settings_current": "config_settings_current",
+    "contract_code_current": "contract_code_current",
+    "contract_data_current": "contract_data_current",
     "liquidity_pools_current": "liquidity_pools_current",
     "offers_current": "offers_current",
-    "trustlines_current": "trust_lines_current"
+    "trustlines_current": "trust_lines_current",
+    "ttl": "ttl"
   },
   "use_testnet": "True",
   "sandbox_dataset": "crypto_stellar_internal_sandbox",

airflow_variables_prod.json

@@ -60,7 +60,7 @@
   "dbt_image_name": "stellar/stellar-dbt:6ee7342",
   "dbt_job_execution_timeout_seconds": 1800,
   "dbt_job_retries": 1,
-  "dbt_mart_dataset": "sdf_marts",
+  "dbt_mart_dataset": "crypto_stellar_dbt",
   "dbt_maximum_bytes_billed": 100000000000000,
   "dbt_project": "hubble-261722",
   "dbt_target": "prod",
@@ -246,9 +246,13 @@
   "dbt_tables": {
     "signers_current": "account_signers_current",
     "accounts_current": "accounts_current",
+    "config_settings_current": "config_settings_current",
+    "contract_code_current": "contract_code_current",
+    "contract_data_current": "contract_data_current",
     "liquidity_pools_current": "liquidity_pools_current",
     "offers_current": "offers_current",
-    "trustlines_current": "trust_lines_current"
+    "trustlines_current": "trust_lines_current",
+    "ttl": "ttl"
   },
   "use_testnet": "False",
   "sandbox_dataset": "crypto_stellar_internal_sandbox",

dags/queries/create_table.sql

@@ -4,4 +4,5 @@ options (partition_expiration_days = 180) as (
     select *
     from `{project_id}.{dataset_id}.{table_id}`
     where batch_run_date >= date_sub(current_date(), interval 6 month)
+    and batch_run_date < date_sub(current_date())
 )

dags/queries/update_table.sql

@@ -1,4 +1,4 @@
 insert into {project_id}.{target_dataset}.{table_id}
 select *
 from {project_id}.{dataset_id}.{table_id}
-where date_trunc(batch_run_date, day) = date_trunc(current_date() - interval 1 day, day)
+where date_trunc(batch_run_date, day) = date_trunc({batch_run_date})

dags/sandbox_create_dag.py

@@ -23,7 +23,6 @@
 with DAG(
     "sandbox_create_dag",
     default_args=get_default_dag_args(),
-    start_date=datetime(2023, 1, 1),
     description="This DAG creates a sandbox",
     schedule_interval="@once",
     params={"alias": "sandbox_dataset"},
@@ -32,8 +31,8 @@
     },
     catchup=False,
 ) as dag:
-    PROJECT = Variable.get("bq_project")
-    DATASET = Variable.get("bq_dataset")
+    PROJECT = Variable.get("public_project")
+    DATASET = Variable.get("public_dataset")
     SANDBOX_DATASET = Variable.get("sandbox_dataset")
     DBT_DATASET = Variable.get("dbt_mart_dataset")
     TABLES_ID = Variable.get("table_ids", deserialize_json=True)

dags/sandbox_update_dag.py

@@ -12,6 +12,7 @@
     file_to_string,
     get_query_filepath,
 )
+from stellar_etl_airflow import macros
 from stellar_etl_airflow.build_cross_dependency_task import build_cross_deps
 from stellar_etl_airflow.default import (
     alert_after_max_retries,
@@ -24,22 +25,28 @@
 with DAG(
     "sandbox_update_dag",
     default_args=get_default_dag_args(),
-    start_date=datetime(2023, 1, 1),
+    start_date=datetime(2024, 4, 30),
     description="This DAG updates a sandbox",
     schedule_interval="0 1 * * *",
     params={"alias": "sandbox_dataset"},
     user_defined_filters={"fromjson": lambda s: loads(s)},
-    catchup=False,
+    catchup=True,
 ) as dag:
     TABLES_ID = Variable.get("table_ids", deserialize_json=True)
     PROJECT = Variable.get("bq_project")
     BQ_DATASET = Variable.get("bq_dataset")
     SANDBOX_DATASET = Variable.get("sandbox_dataset")
 
+    batch_run_date = "{{ batch_run_date_as_datetime_string(dag, data_interval_start) }}"
+
     start_tables_task = EmptyOperator(task_id="start_tables_task")
 
-    wait_on_dag = build_cross_deps(
-        dag, "wait_on_base_tables", "history_archive_with_captive_core_combined_export"
+    wait_on_history_dag = build_cross_deps(
+        dag, "wait_on_base_tables", "history_table_export"
+    )
+
+    wait_on_state_dag = build_cross_deps(
+        dag, "wait_on_base_tables", "state_table_export"
     )
 
     for table_id in TABLES_ID:
@@ -50,6 +57,7 @@
             "dataset_id": BQ_DATASET,
             "table_id": TABLES_ID[table_id],
             "target_dataset": SANDBOX_DATASET,
+            "batch_run_date": batch_run_date
         }
         query = query.format(**sql_params)
         tables_update_task = BigQueryInsertJobOperator(
@@ -63,4 +71,4 @@
             on_failure_callback=alert_after_max_retries,
         )
 
-        start_tables_task >> wait_on_dag >> tables_update_task
+        start_tables_task >> [wait_on_history_dag, wait_on_state_dag] >> tables_update_task