Merge pull request #101 from ajschofield/test/transform-helper-functions

test: transform refactoring - it now loads parquet files into s3 bucket

1 files changed, 28 insertions, 23 deletions

diff --git a/src/dataframes.py b/src/dataframes.py
index ce15872..f122368 100644
--- a/src/dataframes.py
+++ b/src/dataframes.py
@@ -20,22 +20,23 @@ import requests
 def create_fact_sales_order(dict_of_df):
     df_sales = dict_of_df["sales_order"]
     df_sales.index.name = "sales_record_id"
-    df_sales["created_date"] = pd.to_datetime(
-        df_sales["created_at"].dt.date, format="%Y-%m-%d"
+    df_sales["created_date"] = df_sales["created_at"].astype("datetime64[ns]").dt.date
+    df_sales["created_time"] = (
+        df_sales["created_at"].astype("datetime64[ns]").dt.floor("s").dt.time
     )
-    df_sales["created_time"] = df_sales["created_at"].dt.floor("s").dt.time
-    df_sales["last_updated_date"] = pd.to_datetime(
-        df_sales["last_updated"].dt.date, format="%Y-%m-%d"
+    df_sales["last_updated_date"] = (
+        df_sales["last_updated"].astype("datetime64[ns]").dt.date
+    )
+    df_sales["last_updated_time"] = (
+        df_sales["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time
     )
-    df_sales["last_updated_time"] = df_sales["last_updated"].dt.floor("s").dt.time
-
     df_sales["agreed_delivery_date"] = pd.to_datetime(
         df_sales["agreed_delivery_date"], format="%Y-%m-%d"
     )
     df_sales["agreed_payment_date"] = pd.to_datetime(
         df_sales["agreed_payment_date"], format="%Y-%m-%d"
     )
-    df_sales.drop(labels=["created_at", "last_updated"], axis=1, inplace=True)
+    df_sales = df_sales.drop(labels=["created_at", "last_updated"], axis=1)
     df_sales.reset_index(inplace=True)
     return df_sales
 
@@ -46,21 +47,21 @@ def create_fact_sales_order(dict_of_df):
 def create_fact_purchase_orders(dict_of_df):
     df_po = dict_of_df["purchase_order"]
     df_po.index.name = "purchase_record_id"
-    df_po["created_date"] = pd.to_datetime(
-        df_po["created_at"].dt.date, format="%Y-%m-%d"
+    df_po["created_date"] = df_po["created_at"].astype("datetime64[ns]").dt.date
+    df_po["created_time"] = (
+        df_po["created_at"].astype("datetime64[ns]").dt.floor("s").dt.time
     )
-    df_po["created_time"] = df_po["created_at"].dt.floor("s").dt.time
-    df_po["last_updated_date"] = pd.to_datetime(
-        df_po["last_updated"].dt.date, format="%Y-%m-%d"
+    df_po["last_updated_date"] = df_po["last_updated"].astype("datetime64[ns]").dt.date
+    df_po["last_updated_time"] = (
+        df_po["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time
     )
-    df_po["last_updated_time"] = df_po["last_updated"].dt.floor("s").dt.time
     df_po["agreed_delivery_date"] = pd.to_datetime(
         df_po["agreed_delivery_date"], format="%Y-%m-%d"
     )
     df_po["agreed_payment_date"] = pd.to_datetime(
         df_po["agreed_payment_date"], format="%Y-%m-%d"
     )
-    df_po.drop(labels=["created_at", "last_updated"], axis=1, inplace=True)
+    df_po = df_po.drop(labels=["created_at", "last_updated"], axis=1)
     df_po.reset_index(inplace=True)
     return df_po
 
@@ -71,18 +72,22 @@ def create_fact_purchase_orders(dict_of_df):
 def create_fact_payment(dict_of_df):
     df_payment = dict_of_df["payment"]
     df_payment.index.name = "payment_record_id"
-    df_payment["created_date"] = pd.to_datetime(
-        df_payment["created_at"].dt.date, format="%Y-%m-%d"
+    df_payment["created_date"] = (
+        df_payment["created_at"].astype("datetime64[ns]").dt.date
+    )
+    df_payment["created_time"] = (
+        df_payment["created_at"].astype("datetime64[ns]").dt.floor("s").dt.time
+    )
+    df_payment["last_updated_date"] = (
+        df_payment["last_updated"].astype("datetime64[ns]").dt.date
     )
-    df_payment["created_time"] = df_payment["created_at"].dt.floor("s").dt.time
-    df_payment["last_updated_date"] = pd.to_datetime(
-        df_payment["last_updated"].dt.date, format="%Y-%m-%d"
+    df_payment["last_updated_time"] = (
+        df_payment["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time
     )
-    df_payment["last_updated_time"] = df_payment["last_updated"].dt.floor("s").dt.time
     df_payment["payment_date"] = pd.to_datetime(
         df_payment["payment_date"], format="%Y-%m-%d"
     )
-    df_payment.drop(labels=["created_at", "last_updated"], axis=1, inplace=True)
+    df_payment = df_payment.drop(labels=["created_at", "last_updated"], axis=1)
     df_payment.reset_index(inplace=True)
     return df_payment
 
@@ -138,7 +143,7 @@ def create_dim_date(dict_of_df):
     list_of_date_columns = []
     for df in fact_dfs:
         date_col_names = [
-            col_name for col_name in list(df.columns) if "date" in col_name
+            col_name for col_name in list(df.columns) if "_date" in col_name
         ]
         for col in date_col_names:
             list_of_date_columns.append(df[col])