From 7a0c30029bf2f263bd06a1d59a7a34a7979d1dce Mon Sep 17 00:00:00 2001 From: jayckaiser Date: Wed, 29 Jul 2026 11:22:17 -0500 Subject: [PATCH 1/7] First pass at replacing ds/ts with dag_run.start_date equivalents. --- edu_edfi_airflow/callables/change_version.py | 2 +- edu_edfi_airflow/callables/log_util.py | 8 ++++---- edu_edfi_airflow/callables/total_counts.py | 2 +- edu_edfi_airflow/dags/edfi_resource_dag.py | 2 +- .../providers/snowflake/transfers/s3_to_snowflake.py | 2 +- 5 files changed, 8 insertions(+), 8 deletions(-) diff --git a/edu_edfi_airflow/callables/change_version.py b/edu_edfi_airflow/callables/change_version.py index 1659db5c..b9e5e3ba 100644 --- a/edu_edfi_airflow/callables/change_version.py +++ b/edu_edfi_airflow/callables/change_version.py @@ -297,7 +297,7 @@ def update_change_versions( for endpoint in endpoints: row = [ tenant_code, api_year, endpoint, - context["ds"], context["ts"], + context["dag_run"].start_date.date(), context['dag_run'].start_date.isoformat(), edfi_change_version, True, row_set["get_deletes"], ] diff --git a/edu_edfi_airflow/callables/log_util.py b/edu_edfi_airflow/callables/log_util.py index dd625ff1..1b1232e7 100644 --- a/edu_edfi_airflow/callables/log_util.py +++ b/edu_edfi_airflow/callables/log_util.py @@ -94,9 +94,9 @@ def capture_logs_to_snowflake( def wrapper(*args, **kwargs): # Robustly extract the Airflow context - context = kwargs.get("context") or (kwargs if {"ds", "ts"} <= kwargs.keys() else None) + context = kwargs.get("context") or (kwargs if "dag_run" in kwargs else None) if not context: - raise ValueError("Airflow context not found. Cannot extract ds/ts.") + raise ValueError("Airflow context not found. Cannot extract dag_run.start_datel.") def flush_logs(log_records, context): structured = [json.loads(r) for r in log_records] @@ -147,8 +147,8 @@ def flush_logs(log_records, context): api_year=api_year, run_type=run_type, grain_update=grain_update, - run_date=context.get("ds"), - run_timestamp=context.get("ts"), + run_date=context['dag_run'].start_date.date(), + run_timestamp=context['dag_run'].start_date.isoformat(), ) with structured_log_capture(args, kwargs=context) as log_records: diff --git a/edu_edfi_airflow/callables/total_counts.py b/edu_edfi_airflow/callables/total_counts.py index c71e9330..0529e3e8 100644 --- a/edu_edfi_airflow/callables/total_counts.py +++ b/edu_edfi_airflow/callables/total_counts.py @@ -125,7 +125,7 @@ def insert_total_counts( for endpoint in endpoint_counts: row = [ tenant_code, api_year, endpoint[0], - kwargs["ds"], kwargs["ts"], + kwargs["dag_run"].start_date.date(), kwargs['dag_run'].start_date.isoformat(), endpoint[1] ] rows_to_insert.append(row) diff --git a/edu_edfi_airflow/dags/edfi_resource_dag.py b/edu_edfi_airflow/dags/edfi_resource_dag.py index f08c3d06..76bcafbf 100644 --- a/edu_edfi_airflow/dags/edfi_resource_dag.py +++ b/edu_edfi_airflow/dags/edfi_resource_dag.py @@ -261,7 +261,7 @@ def chain_task_groups_into_dag(self): # Set parent directory and create subfolders for each task group. parent_directory = os.path.join( - self.tenant_code, str(self.api_year), "{{ ds_nodash }}", "{{ ts_nodash }}" + self.tenant_code, str(self.api_year), "{{ dag_run.start_date | ds_nodash }}", "{{ dag_run.start_date | ts_nodash }}" ) # Resources diff --git a/edu_edfi_airflow/providers/snowflake/transfers/s3_to_snowflake.py b/edu_edfi_airflow/providers/snowflake/transfers/s3_to_snowflake.py index 855c6b38..095d70b4 100644 --- a/edu_edfi_airflow/providers/snowflake/transfers/s3_to_snowflake.py +++ b/edu_edfi_airflow/providers/snowflake/transfers/s3_to_snowflake.py @@ -248,7 +248,7 @@ def run_bulk_sql_queries(self, table: str, s3_dir: str = '', names: List[str] = Alternative delete and copy queries to be run when all data is sent to the same table in Snowflake. S3 Path Structure: - /{tenant_code}/{api_year}/{ds_nodash}/{ts_no_dash}/{taskgroup_type}/{name}.jsonl + /{tenant_code}/{api_year}/{dag_run.start_date | ds_nodash}/{dag_run.start_date | ts_nodash}/{taskgroup_type}/{name}.jsonl Use regex to capture name: ".+/(\\w+).jsonl?" Note optional args in REGEXP_SUBSTR(): position (1), occurrence (1), regex_parameters ('c'), group_num From 7b319917f7c917b7d9275c4345a1560d345e2bf1 Mon Sep 17 00:00:00 2001 From: jayckaiser Date: Wed, 29 Jul 2026 11:39:28 -0500 Subject: [PATCH 2/7] Revert updates made in log_util (out of scope in deprecated DAG). --- edu_edfi_airflow/callables/log_util.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/edu_edfi_airflow/callables/log_util.py b/edu_edfi_airflow/callables/log_util.py index 1b1232e7..f8af1498 100644 --- a/edu_edfi_airflow/callables/log_util.py +++ b/edu_edfi_airflow/callables/log_util.py @@ -160,4 +160,4 @@ def flush_logs(log_records, context): finally: flush_logs(log_records, context) - return wrapper + return wrapper \ No newline at end of file From 93a1fd56d7c38145ca4d9180a44fe10b5bd17215 Mon Sep 17 00:00:00 2001 From: jayckaiser Date: Wed, 29 Jul 2026 11:43:08 -0500 Subject: [PATCH 3/7] Actual reversion of previous commit. --- edu_edfi_airflow/callables/log_util.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/edu_edfi_airflow/callables/log_util.py b/edu_edfi_airflow/callables/log_util.py index f8af1498..d2cafdc7 100644 --- a/edu_edfi_airflow/callables/log_util.py +++ b/edu_edfi_airflow/callables/log_util.py @@ -94,9 +94,9 @@ def capture_logs_to_snowflake( def wrapper(*args, **kwargs): # Robustly extract the Airflow context - context = kwargs.get("context") or (kwargs if "dag_run" in kwargs else None) + context = kwargs.get("context") or (kwargs if {"ds", "ts"} <= kwargs.keys() else None) if not context: - raise ValueError("Airflow context not found. Cannot extract dag_run.start_datel.") + raise ValueError("Airflow context not found. Cannot extract ds/ts.") def flush_logs(log_records, context): structured = [json.loads(r) for r in log_records] @@ -147,8 +147,8 @@ def flush_logs(log_records, context): api_year=api_year, run_type=run_type, grain_update=grain_update, - run_date=context['dag_run'].start_date.date(), - run_timestamp=context['dag_run'].start_date.isoformat(), + run_date=context.get("ds"), + run_timestamp=context.get("ts"), ) with structured_log_capture(args, kwargs=context) as log_records: From 219a9cbc1d2294fe5450c631668b7137517625f0 Mon Sep 17 00:00:00 2001 From: jayckaiser Date: Wed, 29 Jul 2026 11:45:20 -0500 Subject: [PATCH 4/7] Add missing whitespace to complete revert. --- edu_edfi_airflow/callables/log_util.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/edu_edfi_airflow/callables/log_util.py b/edu_edfi_airflow/callables/log_util.py index d2cafdc7..dd625ff1 100644 --- a/edu_edfi_airflow/callables/log_util.py +++ b/edu_edfi_airflow/callables/log_util.py @@ -160,4 +160,4 @@ def flush_logs(log_records, context): finally: flush_logs(log_records, context) - return wrapper \ No newline at end of file + return wrapper From e4b0fca5f3b802acf7c135a7fddd9ca066f38640 Mon Sep 17 00:00:00 2001 From: jayckaiser Date: Wed, 29 Jul 2026 11:50:15 -0500 Subject: [PATCH 5/7] Revert documentation changes in deprecated provider. --- .../providers/snowflake/transfers/s3_to_snowflake.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/edu_edfi_airflow/providers/snowflake/transfers/s3_to_snowflake.py b/edu_edfi_airflow/providers/snowflake/transfers/s3_to_snowflake.py index 095d70b4..855c6b38 100644 --- a/edu_edfi_airflow/providers/snowflake/transfers/s3_to_snowflake.py +++ b/edu_edfi_airflow/providers/snowflake/transfers/s3_to_snowflake.py @@ -248,7 +248,7 @@ def run_bulk_sql_queries(self, table: str, s3_dir: str = '', names: List[str] = Alternative delete and copy queries to be run when all data is sent to the same table in Snowflake. S3 Path Structure: - /{tenant_code}/{api_year}/{dag_run.start_date | ds_nodash}/{dag_run.start_date | ts_nodash}/{taskgroup_type}/{name}.jsonl + /{tenant_code}/{api_year}/{ds_nodash}/{ts_no_dash}/{taskgroup_type}/{name}.jsonl Use regex to capture name: ".+/(\\w+).jsonl?" Note optional args in REGEXP_SUBSTR(): position (1), occurrence (1), regex_parameters ('c'), group_num From a8d20ed873030b502291870017ef7c04d8040888 Mon Sep 17 00:00:00 2001 From: jayckaiser Date: Wed, 29 Jul 2026 11:54:46 -0500 Subject: [PATCH 6/7] Fix datetype string representation in helpers. --- edu_edfi_airflow/callables/change_version.py | 2 +- edu_edfi_airflow/callables/total_counts.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/edu_edfi_airflow/callables/change_version.py b/edu_edfi_airflow/callables/change_version.py index b9e5e3ba..b7e45640 100644 --- a/edu_edfi_airflow/callables/change_version.py +++ b/edu_edfi_airflow/callables/change_version.py @@ -297,7 +297,7 @@ def update_change_versions( for endpoint in endpoints: row = [ tenant_code, api_year, endpoint, - context["dag_run"].start_date.date(), context['dag_run'].start_date.isoformat(), + context["dag_run"].start_date.date().iso_format(), context['dag_run'].start_date.isoformat(), edfi_change_version, True, row_set["get_deletes"], ] diff --git a/edu_edfi_airflow/callables/total_counts.py b/edu_edfi_airflow/callables/total_counts.py index 0529e3e8..c91830c2 100644 --- a/edu_edfi_airflow/callables/total_counts.py +++ b/edu_edfi_airflow/callables/total_counts.py @@ -125,7 +125,7 @@ def insert_total_counts( for endpoint in endpoint_counts: row = [ tenant_code, api_year, endpoint[0], - kwargs["dag_run"].start_date.date(), kwargs['dag_run'].start_date.isoformat(), + kwargs["dag_run"].start_date.date().isoformat(), kwargs['dag_run'].start_date.isoformat(), endpoint[1] ] rows_to_insert.append(row) From 8c8d265898f740237248b5b226df278d0125ca83 Mon Sep 17 00:00:00 2001 From: jayckaiser Date: Wed, 29 Jul 2026 11:58:04 -0500 Subject: [PATCH 7/7] Bugfix: method mispelling. --- edu_edfi_airflow/callables/change_version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/edu_edfi_airflow/callables/change_version.py b/edu_edfi_airflow/callables/change_version.py index b7e45640..9256dc97 100644 --- a/edu_edfi_airflow/callables/change_version.py +++ b/edu_edfi_airflow/callables/change_version.py @@ -297,7 +297,7 @@ def update_change_versions( for endpoint in endpoints: row = [ tenant_code, api_year, endpoint, - context["dag_run"].start_date.date().iso_format(), context['dag_run'].start_date.isoformat(), + context["dag_run"].start_date.date().isoformat(), context['dag_run'].start_date.isoformat(), edfi_change_version, True, row_set["get_deletes"], ]