Rc/0.3.1 (#46)

* Update metadata to version 0.3.1.

* Hotfix/ea custom dag (#44)

* Add missing default_args argument to EA Custom DAG super init.

* Update CHANGELOG.

* Feature/manual_uploads (#45)

* change date location

* lower db name

* Update s3_to_snowflake_dag.py

* fix copy into with different pathing

* manual upload flag plus date/time regex

* Back to original datalake path

* Fix default source path

* Update CHANGELOG.

---------

Co-authored-by: jayckaiser <jayckaiser@gmail.com>

---------

Co-authored-by: Julianna Alvord <33033206+jalvord1@users.noreply.github.com>

Author: jayckaiser

.gitignore

@@ -0,0 +1,2 @@
+.idea/
+*egg-info*

CHANGELOG.md

@@ -1,3 +1,14 @@
+# ea_airflow_util v0.3.1
+## New features
+- Boolean argument `is_manual_upload` in `S3ToSnowflakeDag` rearranges S3 source pathing to easier structure for partners
+
+## Under the hood
+- Copy statement in `S3ToSnowflakeDag` uses regex instead of string-splitting to infer pull-date and pull-timestamp
+
+## Fixes
+- Fix bug in `EACustomDAG` where `default_args` were not passed to DAG super init.
+
+
 # ea_airflow_util v0.3.0
 ## New features
 - Migrate FTP, ShareFile, casing, and ZIP utilities from Rally into `ea_airflow_util`

ea_airflow_util/dags/ea_custom_dag.py

@@ -55,6 +55,7 @@ def __init__(self,
             catchup=catchup,
             render_template_as_native_obj=render_template_as_native_obj,
             max_active_runs=max_active_runs,
+            default_args=default_args,
             user_defined_macros=user_defined_macros,
             sla_miss_callback=slack_sla_miss_callback,
             **self.subset_kwargs_to_class(DAG, kwargs)  # Remove kwargs not expected in DAG.

ea_airflow_util/dags/s3_to_snowflake_dag.py

@@ -37,6 +37,7 @@ def __init__(self,
         s3_source_conn_id: str,
         s3_dest_conn_id: str,
         s3_dest_file_extension: str,
+        is_manual_upload: bool = False,
 
         pool: str,
         full_replace: bool = False,  #TODO once on latest version of airflow, use dagrun parameter to allow full_replace runs even if not set here at dag level
@@ -59,6 +60,7 @@ def __init__(self,
         self.s3_source_conn_id = s3_source_conn_id
         self.s3_dest_conn_id = s3_dest_conn_id
         self.s3_dest_file_extension = s3_dest_file_extension
+        self.is_manual_upload = is_manual_upload
 
         self.full_replace = full_replace
         self.pool = pool
@@ -70,18 +72,26 @@ def build_s3_to_snowflake_dag(self, **kwargs):
 
         for resource_name in self.resource_names:
 
-            s3_source_prefix = os.path.join(
-                self.tenant_code, self.data_source,
-                str(self.api_year), '{{ ds_nodash }}',
-                resource_name
-            )
-
+            # different source prefix depending on whether the upload to external bucket is manual or not
+            if self.is_manual_upload:
+                s3_source_prefix = os.path.join(
+                    self.tenant_code, self.data_source,
+                    resource_name, str(self.api_year), 
+                    '{{ ds_nodash }}'
+                ) 
+            else:
+                s3_source_prefix = os.path.join(
+                    self.tenant_code, self.data_source,
+                    str(self.api_year), '{{ ds_nodash }}',
+                    resource_name
+                )
+            
             datalake_prefix = os.path.join(
                 self.tenant_code, str(self.api_year),
                 '{{ ds_nodash }}', '{{ ts_nodash }}',
                 resource_name
             )
-            
+
             ## List the s3 files from the source bucket
             list_s3_objects = S3ListOperator(
                 task_id=f'list_s3_objects_{resource_name}',
@@ -155,6 +165,10 @@ def copy_from_datalake_to_raw(self, resource_name, datalake_prefix, full_replace
               and api_year = '{self.api_year}'
         '''
 
+        date_regex = "\\\\d{8}"
+        ts_regex = "\\\\d{8}T\\\\d{6}"
+
+        
         logging.info(f"Copying from data lake to raw: {datalake_prefix}")
         copy_sql = f'''
             copy into {self.database}.{self.schema}.{self.data_source}__{resource_name}
@@ -163,8 +177,8 @@ def copy_from_datalake_to_raw(self, resource_name, datalake_prefix, full_replace
                 select
                     '{self.tenant_code}' as tenant_code,
                     '{self.api_year}' as api_year,
-                    to_date(split_part(metadata$filename, '/', 3), 'YYYYMMDD') as pull_date,
-                    to_timestamp(split_part(metadata$filename, '/', 4), 'YYYYMMDDTHH24MISS') as pull_timestamp,
+                    TO_DATE(REGEXP_SUBSTR(metadata$filename, '{date_regex}'), 'YYYYMMDD') AS pull_date,
+                    TO_TIMESTAMP(REGEXP_SUBSTR(metadata$filename, '{ts_regex}'), 'YYYYMMDDTHH24MISS') AS pull_timestamp,
                     metadata$file_row_number as file_row_number,
                     metadata$filename as filename,
                     '{resource_name}' as name,

setup.py

@@ -6,7 +6,7 @@
 
 setuptools.setup(
       name='ea_airflow_util',
-      version='0.3.0',
+      version='0.3.1',
       description='EA Airflow tools',
       license_files=['LICENSE'],
       url='https://github.com/edanalytics/ea_airflow_util',