Merge branch 'feature/transform-fact-sales-order' of https://github.com/ajschofield/de-project-bentley into feature/transform-fact-sales-order

4 files changed, 62 insertions, 10 deletions

diff --git a/requirements.txt b/requirements.txt
index 62ebbf4..0c81216 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -29,4 +29,6 @@ urllib3==2.2.2
 Werkzeug==3.0.3
 xmltodict==0.13.0
 s3fs 
-pandas
\ No newline at end of file
+pandas
+bs4
+pyarrow
\ No newline at end of file
diff --git a/src/fact-purchase-table.py b/src/fact_purchase_table.py
index 597f104..f1d8fe1 100644
--- a/src/fact-purchase-table.py
+++ b/src/fact_purchase_table.py
@@ -1,3 +1,4 @@
+from bs4 import BeautifulSoup
 from src.transform_lambda import read_from_s3_subfolder_to_df, tables
 from src.extract_lambda import extract_bucket
 import json
@@ -6,7 +7,6 @@ import re
 import pandas as pd
 from datetime import datetime as dt
 import requests
-from bs4 import BeautifulSoup
 
 
 ## dim_staff table is the same across the schemas (no change)
diff --git a/src/transform_lambda.py b/src/transform_lambda.py
index 920a24f..6024a24 100644
--- a/src/transform_lambda.py
+++ b/src/transform_lambda.py
@@ -2,10 +2,11 @@ import json
 import boto3
 import re
 import pandas as pd
-
-
-def lambda_handler(event, context):
-    pass
+import pyarrow as pa
+import pyarrow.parquet as pq
+from src.extract_lambda import extract_bucket
+from src.fact_purchase_table import *
+from src.fact_sales_order import create_dim_staff, create_dim_design, create_fact_sales_order
 
 
 tables = [
@@ -22,6 +23,47 @@ tables = [
     "payment_type",
 ]
 
+def lambda_handler(event, context):
+    dict_of_df = read_from_s3_subfolder_to_df(tables, extract_bucket(), client=boto3.client("s3"))
+    common_df_list = [create_dim_counterparty(dict_of_df), 
+                      create_dim_date(dict_of_df), 
+                      create_dim_location(dict_of_df), 
+                      create_dim_currency(dict_of_df), 
+                      create_dim_staff(dict_of_df)] 
+    
+    create_fact_purchase_order()
+
+    f_sales_list = [create_fact_sales_order(),
+                    create_dim_design()]
+                    
+    
+    '''
+    #dict{
+        sales_schema: {
+            Table_name: df_value, 
+            ...}
+        payment_schema: 
+            Table_name: df_value, 
+            ...}
+        purchase_schema: 
+            Table_name: df_value, 
+            ...}
+    }
+
+    for schema in dict:
+        for table_name, df_value in schema.items():
+            parquet_file = df_value.to_parquet(f'{table_name}.parquet', engine='pyarrow'/'fastparquet'(?)) #we don't know the engine
+
+            s3_key = datetime.strftime(
+                        datetime.today(), f"{schema}/%Y/%m/%d/{table_name}_%H:%M:%S.parquet"
+                    )
+
+            client.upload_file(
+            parquet_file, transform_bucket(), s3_key)
+            ##might need seperate function for easier testing##
+    '''
+
+
 
 def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")):
     table_dfs = {}
@@ -34,4 +76,10 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")):
         table_dfs[table] = pd.concat(list_of_df)
     return table_dfs
 
+def transform_bucket(client=boto3.client("s3")):
+    response = client.list_buckets()
+    bucket_filter = [
+        bucket["Name"] for bucket in response["Buckets"] if "transform" in bucket["Name"]
+    ]
 
+    return bucket_filter[0]
diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py
index 5121905..516f83b 100644
--- a/tests/test_transform_lambda.py
+++ b/tests/test_transform_lambda.py
@@ -39,8 +39,8 @@ class TestReadFromS3:
         )
         print(result)
         expected_df = pd.DataFrame(
-            np.array([["Vegetable", "Sour", "Green"], ["Berry", "Sweet", "Red"]]),
-            columns=["Food_type", "Flavour", "Colour"],
+            np.array([["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"]]),
+            columns=["Food_type", "Flavour", "Colour", "last_updated"],
         )
         assert isinstance(result, dict)
         assert list(result.keys())[0] == "Foods"
@@ -56,8 +56,8 @@ class TestReadFromS3:
             tables, bucket="dummy_buc", client=s3_client
         )
         expected_foods_df = pd.DataFrame(
-            np.array([["Vegetable", "Sour", "Green"], ["Berry", "Sweet", "Red"]]),
-            columns=["Food_type", "Flavour", "Colour"],
+            np.array([["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"]]),
+            columns=["Food_type", "Flavour", "Colour", "last_updated"],
         )
         expected_cars_df = pd.DataFrame(
             np.array(
@@ -72,3 +72,5 @@ class TestReadFromS3:
         assert list(result.keys()) == tables
         assert result["Foods"].eq(expected_foods_df, axis="columns").all(axis=None)
         assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None)
+
+