From 67de54d70ee918bbaf537cb2c119990c4a70c9a7 Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Thu, 22 Aug 2024 16:55:48 +0100
Subject: add convert parquet to df function

---
 src/load_lambda.py | 50 ++++++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 48 insertions(+), 2 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index c6a8e60..2f0c33a 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -1,2 +1,48 @@
-def lambda_handler():
-    pass
+import boto3
+from botocore.exceptions import ClientError
+from pg8000.native import Connection, InterfaceError, identifier
+import pandas as pd
+import pyarrow.parquet as pq
+from io import BytesIO
+
+from botocore.exceptions import ClientError
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+logging.basicConfig(
+    format="{asctime} - {levelname} - {message}",
+    style="{",
+    datefmt="%Y-%m-%d %H:%M",
+    level=logging.DEBUG,
+)
+
+logging.getLogger("botocore").setLevel(logging.WARNING)
+    
+def convert_parquet_files_to_dfs(bucket_name=None, client=None):
+    try:
+        if client is None:
+            client = boto3.client("s3")
+        if bucket_name is None:
+            bucket_name = "transform_bucket"
+        files = client.list_objects_v2(Bucket=bucket_name)
+
+        dfs = []
+        for file in files:
+            file_key = file['Key']
+            try:
+                file_obj = client.get_object(Bucket=bucket_name, Key=file_key)
+                parquet_file = pq.ParquetFile(BytesIO(file_obj['body'].read()))
+                df = parquet_file.read().to_pandas()
+                dfs.append(df)
+            except ClientError as e:
+                logger.error(f"Unable to retrieve S3 object {file_key}: {e}")
+    except ValueError as value_error:
+        logger.error(f"Unable to list objects: {value_error}")
+        raise
+    except ClientError as client_error:
+        logger.error(f"Unable to list objects: {client_error}")
+
+    return dfs 
+ 
\ No newline at end of file
-- 
cgit v1.2.3


From 6bf831c5387408e92a63cb5667aab8f415b536e4 Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Fri, 23 Aug 2024 09:40:08 +0100
Subject: add improved convert parquet files to df function

---
 src/load_lambda.py | 33 +++++++++++++++++++--------------
 1 file changed, 19 insertions(+), 14 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 2f0c33a..1813db4 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -1,11 +1,8 @@
 import boto3
 from botocore.exceptions import ClientError
-from pg8000.native import Connection, InterfaceError, identifier
 import pandas as pd
 import pyarrow.parquet as pq
 from io import BytesIO
-
-from botocore.exceptions import ClientError
 import logging
 
 
@@ -19,7 +16,9 @@ logging.basicConfig(
 )
 
 logging.getLogger("botocore").setLevel(logging.WARNING)
-    
+
+# list and then retrieve parquet files from S3 bucket
+# convert parquet files into dataframes and return a list of dataframes  
 def convert_parquet_files_to_dfs(bucket_name=None, client=None):
     try:
         if client is None:
@@ -29,20 +28,26 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
         files = client.list_objects_v2(Bucket=bucket_name)
 
         dfs = []
-        for file in files:
-            file_key = file['Key']
-            try:
-                file_obj = client.get_object(Bucket=bucket_name, Key=file_key)
-                parquet_file = pq.ParquetFile(BytesIO(file_obj['body'].read()))
-                df = parquet_file.read().to_pandas()
-                dfs.append(df)
-            except ClientError as e:
-                logger.error(f"Unable to retrieve S3 object {file_key}: {e}")
+        if "Contents" in files:
+            for file in files["Contents"]:
+                file_key = file['Key']
+                try:
+                    file_obj = client.get_object(Bucket=bucket_name, Key=file_key)
+                    parquet_file = pq.ParquetFile(BytesIO(file_obj['Body'].read()))
+                    df = parquet_file.read().to_pandas()
+                    dfs.append(df)
+                except ClientError as e:
+                    logger.error(f"Unable to retrieve S3 object {file_key}: {e}")
+                except Exception as e:
+                    logger.error(f"Unable to process file {file_key}: {e}")
+        else:
+            logger.error(f"No files found in {bucket_name}.")
+            return []
     except ValueError as value_error:
         logger.error(f"Unable to list objects: {value_error}")
         raise
     except ClientError as client_error:
         logger.error(f"Unable to list objects: {client_error}")
+        raise
 
     return dfs 
- 
\ No newline at end of file
-- 
cgit v1.2.3


From 265d61c34c3a56b7e74333911e65d3148b2945b4 Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Fri, 23 Aug 2024 09:47:52 +0100
Subject: add get transform bucket function

---
 src/load_lambda.py | 16 +++++++++++++++-
 1 file changed, 15 insertions(+), 1 deletion(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 1813db4..a3fd996 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -17,6 +17,20 @@ logging.basicConfig(
 
 logging.getLogger("botocore").setLevel(logging.WARNING)
 
+# get transform bucket
+def transform_bucket(client=None):
+    if client is None:
+        client = boto3.client("s3")
+    response = client.list_buckets()
+    transform_bucket_filter = [
+        bucket["Name"] for bucket in response["Buckets"] if "transform" in bucket["Name"]
+    ]
+
+    if not transform_bucket_filter:
+        raise ValueError("No transform_bucket found")
+
+    return transform_bucket_filter[0]
+
 # list and then retrieve parquet files from S3 bucket
 # convert parquet files into dataframes and return a list of dataframes  
 def convert_parquet_files_to_dfs(bucket_name=None, client=None):
@@ -24,7 +38,7 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
         if client is None:
             client = boto3.client("s3")
         if bucket_name is None:
-            bucket_name = "transform_bucket"
+            bucket_name = transform_bucket(client)
         files = client.list_objects_v2(Bucket=bucket_name)
 
         dfs = []
-- 
cgit v1.2.3


From 09c8191ce983e4335cfb131d21ddb5413b849cfb Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Fri, 23 Aug 2024 11:18:24 +0100
Subject: add tests

---
 src/load_lambda.py        | 61 ++++++++++++++++++++++++++++++++++++++++++++---
 tests/test_load_lambda.py |  3 +--
 2 files changed, 59 insertions(+), 5 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index a3fd996..d95c27a 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -4,6 +4,9 @@ import pandas as pd
 import pyarrow.parquet as pq
 from io import BytesIO
 import logging
+import json
+from src.extract_lambda import retrieve_secrets, connect_to_database
+from sqlalchemy import create_engine
 
 
 logger = logging.getLogger(__name__)
@@ -17,6 +20,43 @@ logging.basicConfig(
 
 logging.getLogger("botocore").setLevel(logging.WARNING)
 
+def lambda_handler(event, context):
+    db = None
+    try:
+        uploaded_tables = upload_dfs_to_database()
+        if uploaded_tables == []:
+            return {
+                "statusCode": 200,
+                "body": json.dumps("No datframes were uploaded."),
+            }
+        return {
+            "statusCode": 200,
+            "body": json.dumps(
+                f"""The following dataframes were uploaded successfully: 
+                {', '.join(upload_dfs_to_database['updated'])}."""
+            ),
+        }
+    except Exception as e:
+        logger.error(f"Error: {e}", exc_info=True)
+        return {"statusCode": 500, "body": json.dumps("Internal server error.")}
+    finally:
+        if db:
+            db.close()
+
+# connect to database, slightly different way of doing it, to allow manipulation through pandas
+def connect_to_db_and_return_engine():
+    secrets = json.loads(retrieve_secrets("bentley-RDS-credentials"))  #need to amend retrieve secrets function
+    host = secrets["host"]
+    port = secrets["port"]
+    user = secrets["user"]
+    password = secrets["password"]
+    database = secrets["database"]
+    conn_str = f'postgresql+pg8000://{user}:{password}@{host}:{port}/{database}'
+    engine = create_engine(conn_str) #interface between python (pandas) and SQL
+    return engine
+
+
+
 # get transform bucket
 def transform_bucket(client=None):
     if client is None:
@@ -41,7 +81,7 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
             bucket_name = transform_bucket(client)
         files = client.list_objects_v2(Bucket=bucket_name)
 
-        dfs = []
+        dfs = {}
         if "Contents" in files:
             for file in files["Contents"]:
                 file_key = file['Key']
@@ -49,7 +89,7 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
                     file_obj = client.get_object(Bucket=bucket_name, Key=file_key)
                     parquet_file = pq.ParquetFile(BytesIO(file_obj['Body'].read()))
                     df = parquet_file.read().to_pandas()
-                    dfs.append(df)
+                    dfs[file_key] = df
                 except ClientError as e:
                     logger.error(f"Unable to retrieve S3 object {file_key}: {e}")
                 except Exception as e:
@@ -64,4 +104,19 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
         logger.error(f"Unable to list objects: {client_error}")
         raise
 
-    return dfs 
+    return dfs
+
+def upload_dfs_to_database():
+    uploaded = []
+    dict_of_dfs = convert_parquet_files_to_dfs()
+    db_engine = connect_to_db_and_return_engine()
+    try:
+        for table_name, df in dict_of_dfs:
+            df.to_sql(table_name, con=db_engine, ifexists="replace", index=False)
+            uploaded.append(table_name)
+    except Exception as e:
+        logger.error(f"Error uploading dataframes: {e}")
+    db_engine.dispose()
+    return uploaded
+
+    # aiming to return a list of uploaded tables
\ No newline at end of file
diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py
index 0572340..d9ea918 100644
--- a/tests/test_load_lambda.py
+++ b/tests/test_load_lambda.py
@@ -1,8 +1,7 @@
-import boto3
 import pandas as pd
 import pyarrow.parquet as pq
 from io import BytesIO
-from src.load_lambda import convert_parquet_files_to_dataframes
+from src.load_lambda import convert_parquet_files_to_dfs
 
 class TestConvertParquetToDFs:
     def test_convert_parquet_to_dfs_returns_df():
-- 
cgit v1.2.3


From 65289cdd17359c6a29560339e134e0ddf9461ce0 Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Fri, 23 Aug 2024 12:08:09 +0100
Subject: add amendments to load lambda

---
 src/load_lambda.py | 66 ++++++++++++++++++++++++++++++------------------------
 1 file changed, 37 insertions(+), 29 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index d95c27a..f92bb45 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -1,11 +1,11 @@
 import boto3
-from botocore.exceptions import ClientError
+from botocore.exceptions import ClientError, InterfaceError
 import pandas as pd
 import pyarrow.parquet as pq
 from io import BytesIO
 import logging
 import json
-from src.extract_lambda import retrieve_secrets, connect_to_database
+from src.extract_lambda import retrieve_secrets
 from sqlalchemy import create_engine
 
 
@@ -18,67 +18,74 @@ logging.basicConfig(
     level=logging.DEBUG,
 )
 
-logging.getLogger("botocore").setLevel(logging.WARNING)
+logging.getLogger("botocore").setLevel(logging.INFO)
+
 
 def lambda_handler(event, context):
-    db = None
     try:
         uploaded_tables = upload_dfs_to_database()
-        if uploaded_tables == []:
+        if not uploaded_tables:
             return {
                 "statusCode": 200,
-                "body": json.dumps("No datframes were uploaded."),
+                "body": json.dumps("No dataframes were uploaded."),
             }
         return {
             "statusCode": 200,
             "body": json.dumps(
                 f"""The following dataframes were uploaded successfully: 
-                {', '.join(upload_dfs_to_database['updated'])}."""
+                {', '.join(uploaded_tables)} ."""
             ),
         }
     except Exception as e:
         logger.error(f"Error: {e}", exc_info=True)
         return {"statusCode": 500, "body": json.dumps("Internal server error.")}
-    finally:
-        if db:
-            db.close()
 
 # connect to database, slightly different way of doing it, to allow manipulation through pandas
 def connect_to_db_and_return_engine():
-    secrets = json.loads(retrieve_secrets("bentley-RDS-credentials"))  #need to amend retrieve secrets function
-    host = secrets["host"]
-    port = secrets["port"]
-    user = secrets["user"]
-    password = secrets["password"]
-    database = secrets["database"]
-    conn_str = f'postgresql+pg8000://{user}:{password}@{host}:{port}/{database}'
-    engine = create_engine(conn_str) #interface between python (pandas) and SQL
-    return engine
-
-
+    try:
+        secrets = json.loads(retrieve_secrets("bentley-RDS-credentials"))  #need to amend retrieve secrets function
+        host = secrets["host"]
+        port = secrets["port"]
+        user = secrets["user"]
+        password = secrets["password"]
+        database = secrets["database"]
+        conn_str = f'postgresql+pg8000://{user}:{password}@{host}:{port}/{database}'
+        engine = create_engine(conn_str) #interface between python (pandas) and SQL
+        return engine
+    except Exception as e:
+        logger.error(f"Interface error: {e}")
+        raise RuntimeError("Failed to create database engine")
+    
 
 # get transform bucket
-def transform_bucket(client=None):
+def get_transform_bucket(client=None):
     if client is None:
         client = boto3.client("s3")
-    response = client.list_buckets()
+    try:
+        response = client.list_buckets()
+    except ClientError as e:
+        logger.error(f"Error listing S3 buckets: {e}")
+        raise RuntimeError("Error listing S3 buckets")
+    
     transform_bucket_filter = [
         bucket["Name"] for bucket in response["Buckets"] if "transform" in bucket["Name"]
     ]
 
     if not transform_bucket_filter:
-        raise ValueError("No transform_bucket found")
+        logger.error("No transform bucket found")
+        raise ValueError("No transform bucket found")
 
     return transform_bucket_filter[0]
 
 # list and then retrieve parquet files from S3 bucket
-# convert parquet files into dataframes and return a list of dataframes  
+# convert parquet files into dataframes
+# return a dictionary of dataframes with name as key, and dataframe object as value 
 def convert_parquet_files_to_dfs(bucket_name=None, client=None):
     try:
         if client is None:
             client = boto3.client("s3")
         if bucket_name is None:
-            bucket_name = transform_bucket(client)
+            bucket_name = get_transform_bucket()
         files = client.list_objects_v2(Bucket=bucket_name)
 
         dfs = {}
@@ -96,7 +103,7 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
                     logger.error(f"Unable to process file {file_key}: {e}")
         else:
             logger.error(f"No files found in {bucket_name}.")
-            return []
+            return {}
     except ValueError as value_error:
         logger.error(f"Unable to list objects: {value_error}")
         raise
@@ -111,11 +118,12 @@ def upload_dfs_to_database():
     dict_of_dfs = convert_parquet_files_to_dfs()
     db_engine = connect_to_db_and_return_engine()
     try:
-        for table_name, df in dict_of_dfs:
-            df.to_sql(table_name, con=db_engine, ifexists="replace", index=False)
+        for table_name, df in dict_of_dfs.items():
+            df.to_sql(table_name, con=db_engine, if_exists="replace", index=False)
             uploaded.append(table_name)
     except Exception as e:
         logger.error(f"Error uploading dataframes: {e}")
+        raise
     db_engine.dispose()
     return uploaded
 
-- 
cgit v1.2.3


From f3bb705a31ab9d94dc856c2de0da4b7b73a57fae Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Fri, 23 Aug 2024 12:38:25 +0100
Subject: add get transform bucket test

---
 src/load_lambda.py        |  2 +-
 tests/test_load_lambda.py | 48 +++++++++++++++++++++++++++++++++++++++++++----
 2 files changed, 45 insertions(+), 5 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index f92bb45..a9d5ac5 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -1,5 +1,5 @@
 import boto3
-from botocore.exceptions import ClientError, InterfaceError
+from botocore.exceptions import ClientError
 import pandas as pd
 import pyarrow.parquet as pq
 from io import BytesIO
diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py
index d9ea918..2392f10 100644
--- a/tests/test_load_lambda.py
+++ b/tests/test_load_lambda.py
@@ -1,8 +1,48 @@
 import pandas as pd
 import pyarrow.parquet as pq
 from io import BytesIO
-from src.load_lambda import convert_parquet_files_to_dfs
+from moto import mock_aws
+import boto3
+import os
+import pytest
+from src.load_lambda import lambda_handler, connect_to_db_and_return_engine, get_transform_bucket, convert_parquet_files_to_dfs, upload_dfs_to_database
 
-class TestConvertParquetToDFs:
-    def test_convert_parquet_to_dfs_returns_df():
-        
\ No newline at end of file
+@pytest.fixture(scope="class")
+def aws_credentials():
+    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
+    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
+    os.environ["AWS_SECURIT_TOKEN"] = "testing"
+    os.environ["AWS_SESSION_TOKEN"] = "testing"
+    os.environ["AWS_DEFAULT_REGION"] = "eu-west-2"
+
+
+@pytest.fixture(scope="class")
+def s3_client(aws_credentials):
+    with mock_aws():
+        yield boto3.client("s3")
+
+@pytest.fixture(scope="function")
+def s3_mock_bucket(s3_client):
+    bucket = s3_client.create_bucket(
+        Bucket="transform_bucket",
+        CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
+    )
+    return bucket
+
+
+class TestLambdaHandler:
+    pass
+
+class TestConnectToDBAndReturnEngine:
+    pass
+
+class TestGetTransformBucket:
+    def test_get_transform_bucket_returns_string(self, s3_client, s3_mock_bucket):
+        result = get_transform_bucket(s3_client)
+        assert result == "transform_bucket"
+
+class TestConvertParquetToDfs:
+        pass
+
+class TestUploadDfsToDatabase:
+    pass
\ No newline at end of file
-- 
cgit v1.2.3


From 0f8f376fe806ea72f056356cc043213f61159697 Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Fri, 23 Aug 2024 14:35:36 +0100
Subject: add retrieve secrets function

---
 src/load_lambda.py | 21 ++++++++++++++++++++-
 1 file changed, 20 insertions(+), 1 deletion(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index a9d5ac5..2dc90ba 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -40,10 +40,29 @@ def lambda_handler(event, context):
         logger.error(f"Error: {e}", exc_info=True)
         return {"statusCode": 500, "body": json.dumps("Internal server error.")}
 
+def retrieve_secrets():
+    secret_name = "bentley-RDS-credentials"
+    region_name = "eu-west-2"
+
+    # Create a Secrets Manager client
+    session = boto3.session.Session()
+    client = session.client(service_name="secretsmanager", region_name=region_name)
+
+    try:
+        get_secret_value_response = client.get_secret_value(SecretId=secret_name)
+    except ClientError as e:
+        logger.error(f"Failed to retrieve secret {secret_name}: {str(e)}")
+        raise e
+    except KeyError:
+        logger.error(f"Secret {secret_name} does not contain a SecretString")
+        raise ValueError(f"Secret {secret_name} does not contain a SecretString")
+
+    return get_secret_value_response["SecretString"]
+
 # connect to database, slightly different way of doing it, to allow manipulation through pandas
 def connect_to_db_and_return_engine():
     try:
-        secrets = json.loads(retrieve_secrets("bentley-RDS-credentials"))  #need to amend retrieve secrets function
+        secrets = json.loads(retrieve_secrets())
         host = secrets["host"]
         port = secrets["port"]
         user = secrets["user"]
-- 
cgit v1.2.3


From 500ebf24c746ec87c9c846f5a82d638cc23983b9 Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Fri, 23 Aug 2024 17:04:08 +0100
Subject: add amendendments for upload_dfs_to_db

---
 src/load_lambda.py | 47 ++++++++++++++++++++++++++++++++++-------------
 1 file changed, 34 insertions(+), 13 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 2dc90ba..8eaea32 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -24,7 +24,7 @@ logging.getLogger("botocore").setLevel(logging.INFO)
 def lambda_handler(event, context):
     try:
         uploaded_tables = upload_dfs_to_database()
-        if not uploaded_tables:
+        if not uploaded_tables["uploaded"]:
             return {
                 "statusCode": 200,
                 "body": json.dumps("No dataframes were uploaded."),
@@ -33,7 +33,7 @@ def lambda_handler(event, context):
             "statusCode": 200,
             "body": json.dumps(
                 f"""The following dataframes were uploaded successfully: 
-                {', '.join(uploaded_tables)} ."""
+                {uploaded_tables["uploaded"]} ."""
             ),
         }
     except Exception as e:
@@ -133,17 +133,38 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
     return dfs
 
 def upload_dfs_to_database():
-    uploaded = []
+    upload_status = {"uploaded": [], "not_uploaded": []}
     dict_of_dfs = convert_parquet_files_to_dfs()
     db_engine = connect_to_db_and_return_engine()
-    try:
-        for table_name, df in dict_of_dfs.items():
-            df.to_sql(table_name, con=db_engine, if_exists="replace", index=False)
-            uploaded.append(table_name)
-    except Exception as e:
-        logger.error(f"Error uploading dataframes: {e}")
-        raise
+    immutable_df_dict = ["dim_counterparty.parquet", 
+                         "dim_date.parquet", #this needs to be mutable
+                         "dim_location.parquet", 
+                         "dim_staff.parquet", 
+                         "dim_design.parquet"]
+    mutable_df_dict = ["fact_sales_order", 
+                       "fact_purchase_order",  
+                       "fact_payment", 
+                       "dim_currency"]
+    
+    for file_name, df in dict_of_dfs.items():
+        if file_name in immutable_df_dict:
+            table_name = file_name.split(".")[0]
+            try:
+                df.to_sql(table_name, con=db_engine, schema="project_team_2", if_exists="overwrite", index=False)
+                upload_status["uploaded"].append(table_name)
+            except Exception as e:
+                logger.error(f"Error uploading dataframe {file_name} to database: {e}")
+                raise
+        elif file_name.rsplit('_', 1)[0] in mutable_df_dict:
+            table_name = file_name.rsplit('_', 1)[0]
+            try:
+                    df.to_sql(table_name, con=db_engine, schema="project_team_2", if_exists="overwrite", index=False)
+                    upload_status["uploaded"].append(table_name)
+            except Exception as e:
+                logger.error(f"Error uploading dataframe {file_name} to database: {e}")
+                raise
+        else:
+            upload_status["not_uploaded"].append(file_name)
+            logger.error(f"{file_name} does not correspond with table in database")
     db_engine.dispose()
-    return uploaded
-
-    # aiming to return a list of uploaded tables
\ No newline at end of file
+    return upload_status
\ No newline at end of file
-- 
cgit v1.2.3


From 69edb14dad584d45fa6a83a90c08292b84795507 Mon Sep 17 00:00:00 2001
From: "deepsource-autofix[bot]"
 <62050782+deepsource-autofix[bot]@users.noreply.github.com>
Date: Fri, 23 Aug 2024 16:11:45 +0000
Subject: style: format code with Autopep8, Black and Ruff Formatter

This commit fixes the style issues introduced in 0ff2956 according to the output
from Autopep8, Black and Ruff Formatter.

Details: https://github.com/ajschofield/de-project-bentley/pull/95
---
 src/load_lambda.py        | 75 ++++++++++++++++++++++++++++++++---------------
 tests/test_load_lambda.py | 44 +++++++++++++++++----------
 2 files changed, 80 insertions(+), 39 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 8eaea32..6e6bc80 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -40,6 +40,7 @@ def lambda_handler(event, context):
         logger.error(f"Error: {e}", exc_info=True)
         return {"statusCode": 500, "body": json.dumps("Internal server error.")}
 
+
 def retrieve_secrets():
     secret_name = "bentley-RDS-credentials"
     region_name = "eu-west-2"
@@ -59,7 +60,10 @@ def retrieve_secrets():
 
     return get_secret_value_response["SecretString"]
 
+
 # connect to database, slightly different way of doing it, to allow manipulation through pandas
+
+
 def connect_to_db_and_return_engine():
     try:
         secrets = json.loads(retrieve_secrets())
@@ -68,13 +72,14 @@ def connect_to_db_and_return_engine():
         user = secrets["user"]
         password = secrets["password"]
         database = secrets["database"]
-        conn_str = f'postgresql+pg8000://{user}:{password}@{host}:{port}/{database}'
-        engine = create_engine(conn_str) #interface between python (pandas) and SQL
+        conn_str = f"postgresql+pg8000://{user}:{password}@{host}:{port}/{database}"
+        # interface between python (pandas) and SQL
+        engine = create_engine(conn_str)
         return engine
     except Exception as e:
         logger.error(f"Interface error: {e}")
         raise RuntimeError("Failed to create database engine")
-    
+
 
 # get transform bucket
 def get_transform_bucket(client=None):
@@ -85,9 +90,11 @@ def get_transform_bucket(client=None):
     except ClientError as e:
         logger.error(f"Error listing S3 buckets: {e}")
         raise RuntimeError("Error listing S3 buckets")
-    
+
     transform_bucket_filter = [
-        bucket["Name"] for bucket in response["Buckets"] if "transform" in bucket["Name"]
+        bucket["Name"]
+        for bucket in response["Buckets"]
+        if "transform" in bucket["Name"]
     ]
 
     if not transform_bucket_filter:
@@ -96,9 +103,12 @@ def get_transform_bucket(client=None):
 
     return transform_bucket_filter[0]
 
+
 # list and then retrieve parquet files from S3 bucket
 # convert parquet files into dataframes
-# return a dictionary of dataframes with name as key, and dataframe object as value 
+# return a dictionary of dataframes with name as key, and dataframe object as value
+
+
 def convert_parquet_files_to_dfs(bucket_name=None, client=None):
     try:
         if client is None:
@@ -110,10 +120,10 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
         dfs = {}
         if "Contents" in files:
             for file in files["Contents"]:
-                file_key = file['Key']
+                file_key = file["Key"]
                 try:
                     file_obj = client.get_object(Bucket=bucket_name, Key=file_key)
-                    parquet_file = pq.ParquetFile(BytesIO(file_obj['Body'].read()))
+                    parquet_file = pq.ParquetFile(BytesIO(file_obj["Body"].read()))
                     df = parquet_file.read().to_pandas()
                     dfs[file_key] = df
                 except ClientError as e:
@@ -132,34 +142,51 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
 
     return dfs
 
+
 def upload_dfs_to_database():
     upload_status = {"uploaded": [], "not_uploaded": []}
     dict_of_dfs = convert_parquet_files_to_dfs()
     db_engine = connect_to_db_and_return_engine()
-    immutable_df_dict = ["dim_counterparty.parquet", 
-                         "dim_date.parquet", #this needs to be mutable
-                         "dim_location.parquet", 
-                         "dim_staff.parquet", 
-                         "dim_design.parquet"]
-    mutable_df_dict = ["fact_sales_order", 
-                       "fact_purchase_order",  
-                       "fact_payment", 
-                       "dim_currency"]
-    
+    immutable_df_dict = [
+        "dim_counterparty.parquet",
+        "dim_date.parquet",  # this needs to be mutable
+        "dim_location.parquet",
+        "dim_staff.parquet",
+        "dim_design.parquet",
+    ]
+    mutable_df_dict = [
+        "fact_sales_order",
+        "fact_purchase_order",
+        "fact_payment",
+        "dim_currency",
+    ]
+
     for file_name, df in dict_of_dfs.items():
         if file_name in immutable_df_dict:
             table_name = file_name.split(".")[0]
             try:
-                df.to_sql(table_name, con=db_engine, schema="project_team_2", if_exists="overwrite", index=False)
+                df.to_sql(
+                    table_name,
+                    con=db_engine,
+                    schema="project_team_2",
+                    if_exists="overwrite",
+                    index=False,
+                )
                 upload_status["uploaded"].append(table_name)
             except Exception as e:
                 logger.error(f"Error uploading dataframe {file_name} to database: {e}")
                 raise
-        elif file_name.rsplit('_', 1)[0] in mutable_df_dict:
-            table_name = file_name.rsplit('_', 1)[0]
+        elif file_name.rsplit("_", 1)[0] in mutable_df_dict:
+            table_name = file_name.rsplit("_", 1)[0]
             try:
-                    df.to_sql(table_name, con=db_engine, schema="project_team_2", if_exists="overwrite", index=False)
-                    upload_status["uploaded"].append(table_name)
+                df.to_sql(
+                    table_name,
+                    con=db_engine,
+                    schema="project_team_2",
+                    if_exists="overwrite",
+                    index=False,
+                )
+                upload_status["uploaded"].append(table_name)
             except Exception as e:
                 logger.error(f"Error uploading dataframe {file_name} to database: {e}")
                 raise
@@ -167,4 +194,4 @@ def upload_dfs_to_database():
             upload_status["not_uploaded"].append(file_name)
             logger.error(f"{file_name} does not correspond with table in database")
     db_engine.dispose()
-    return upload_status
\ No newline at end of file
+    return upload_status
diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py
index e04ccec..88c71e4 100644
--- a/tests/test_load_lambda.py
+++ b/tests/test_load_lambda.py
@@ -5,7 +5,14 @@ from moto import mock_aws
 import boto3
 import os
 import pytest
-from src.load_lambda import lambda_handler, connect_to_db_and_return_engine, get_transform_bucket, convert_parquet_files_to_dfs, upload_dfs_to_database
+from src.load_lambda import (
+    lambda_handler,
+    connect_to_db_and_return_engine,
+    get_transform_bucket,
+    convert_parquet_files_to_dfs,
+    upload_dfs_to_database,
+)
+
 
 @pytest.fixture(scope="class")
 def aws_credentials():
@@ -25,12 +32,15 @@ def mock_s3_client(aws_credentials):
 class TestLambdaHandler:
     pass
 
+
 class TestRetrieveSecrets:
     pass
 
+
 class TestConnectToDBAndReturnEngine:
     pass
 
+
 class TestGetTransformBucket:
     def test_raises_value_error_if_no_buckets(self, mock_s3_client):
         with pytest.raises(ValueError, match="No transform bucket found"):
@@ -38,35 +48,38 @@ class TestGetTransformBucket:
 
     def test_raises_value_error_if_no_transform_bucket(self, mock_s3_client):
         mock_s3_client.create_bucket(
-        Bucket="extract_bucket",
-        CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
-    )
+            Bucket="extract_bucket",
+            CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
+        )
         with pytest.raises(ValueError, match="No transform bucket found"):
             get_transform_bucket(mock_s3_client)
 
     def test_returns_transform_bucket_if_one_bucket(self, mock_s3_client):
         mock_s3_client.create_bucket(
-        Bucket="transform_bucket",
-        CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
-    )
+            Bucket="transform_bucket",
+            CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
+        )
         result = get_transform_bucket(mock_s3_client)
         assert result == "transform_bucket"
 
     def test_only_returns_transform_bucket_if_several_buckets(self, mock_s3_client):
         mock_s3_client.create_bucket(
-        Bucket="another_test_bucket",
-        CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
-    )
+            Bucket="another_test_bucket",
+            CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
+        )
         result = get_transform_bucket(mock_s3_client)
         assert result == "transform_bucket"
 
+
 class TestConvertParquetToDfs:
     def test_function_returns_empty_dictionary_if_no_files(self, mock_s3_client):
         mock_s3_client.create_bucket(
-        Bucket="transform_bucket",
-        CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
-    )
-        result = convert_parquet_files_to_dfs(bucket_name="transform_bucket", client=mock_s3_client)
+            Bucket="transform_bucket",
+            CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
+        )
+        result = convert_parquet_files_to_dfs(
+            bucket_name="transform_bucket", client=mock_s3_client
+        )
         assert result == {}
 
     # def test_function_returns_dictionary_with_table_with_file_key():
@@ -74,5 +87,6 @@ class TestConvertParquetToDfs:
     #     result = convert_parquet_files_to_dfs(bucket_name="transform_bucket", client=mock_s3_client)
     #     assert "dim_staff" in result
 
+
 class TestUploadDfsToDatabase:
-    pass
\ No newline at end of file
+    pass
-- 
cgit v1.2.3


From 151429859bca904cbacf18f4b169f1f768fa212a Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Tue, 27 Aug 2024 12:01:53 +0100
Subject: remove import as not required

---
 src/load_lambda.py | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 6e6bc80..685c562 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -5,7 +5,6 @@ import pyarrow.parquet as pq
 from io import BytesIO
 import logging
 import json
-from src.extract_lambda import retrieve_secrets
 from sqlalchemy import create_engine
 
 
@@ -169,7 +168,7 @@ def upload_dfs_to_database():
                     table_name,
                     con=db_engine,
                     schema="project_team_2",
-                    if_exists="overwrite",
+                    if_exists="append",
                     index=False,
                 )
                 upload_status["uploaded"].append(table_name)
@@ -183,7 +182,7 @@ def upload_dfs_to_database():
                     table_name,
                     con=db_engine,
                     schema="project_team_2",
-                    if_exists="overwrite",
+                    if_exists="append",
                     index=False,
                 )
                 upload_status["uploaded"].append(table_name)
@@ -195,3 +194,6 @@ def upload_dfs_to_database():
             logger.error(f"{file_name} does not correspond with table in database")
     db_engine.dispose()
     return upload_status
+
+if __name__ == "__main__":
+    lambda_handler(None, None)
-- 
cgit v1.2.3


From 8cd9edde84f4ca706ad93b143c5ff7e3397ce981 Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Tue, 27 Aug 2024 12:28:58 +0100
Subject: add json.loads to retrieve secrests function

---
 src/load_lambda.py | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 685c562..f08e335 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -40,16 +40,19 @@ def lambda_handler(event, context):
         return {"statusCode": 500, "body": json.dumps("Internal server error.")}
 
 
-def retrieve_secrets():
-    secret_name = "bentley-RDS-credentials"
+def retrieve_secrets(client=None, secret_name=None):
+    session = boto3.session.Session()
     region_name = "eu-west-2"
 
-    # Create a Secrets Manager client
-    session = boto3.session.Session()
-    client = session.client(service_name="secretsmanager", region_name=region_name)
+    if secret_name == None:
+        secret_name = "bentley-RDS-credentials"
+    if client == None:
+        client = session.client(service_name="secretsmanager", region_name=region_name)
+
 
     try:
         get_secret_value_response = client.get_secret_value(SecretId=secret_name)
+        print(get_secret_value_response)
     except ClientError as e:
         logger.error(f"Failed to retrieve secret {secret_name}: {str(e)}")
         raise e
@@ -57,7 +60,7 @@ def retrieve_secrets():
         logger.error(f"Secret {secret_name} does not contain a SecretString")
         raise ValueError(f"Secret {secret_name} does not contain a SecretString")
 
-    return get_secret_value_response["SecretString"]
+    return json.loads(get_secret_value_response["SecretString"])
 
 
 # connect to database, slightly different way of doing it, to allow manipulation through pandas
-- 
cgit v1.2.3


From d623c42a891f2fe8a26493354af0d9e299f3c526 Mon Sep 17 00:00:00 2001
From: Alex Schofield <git@ajschof.me>
Date: Tue, 27 Aug 2024 15:19:14 +0100
Subject: refactor: add parameter for sm_secret

---
 src/load_lambda.py | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index f08e335..11d1d70 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -49,7 +49,6 @@ def retrieve_secrets(client=None, secret_name=None):
     if client == None:
         client = session.client(service_name="secretsmanager", region_name=region_name)
 
-
     try:
         get_secret_value_response = client.get_secret_value(SecretId=secret_name)
         print(get_secret_value_response)
@@ -66,9 +65,12 @@ def retrieve_secrets(client=None, secret_name=None):
 # connect to database, slightly different way of doing it, to allow manipulation through pandas
 
 
-def connect_to_db_and_return_engine():
+def connect_to_db_and_return_engine(sm_secret=None):
+    if sm_secret is None:
+        sm_secret = retrieve_secrets()
+
     try:
-        secrets = json.loads(retrieve_secrets())
+        secrets = json.loads(sm_secret)
         host = secrets["host"]
         port = secrets["port"]
         user = secrets["user"]
@@ -198,5 +200,6 @@ def upload_dfs_to_database():
     db_engine.dispose()
     return upload_status
 
+
 if __name__ == "__main__":
     lambda_handler(None, None)
-- 
cgit v1.2.3


From cbfc98a9f43b5a0dae95337057c18c9dc2a298e3 Mon Sep 17 00:00:00 2001
From: Alex Schofield <git@ajschof.me>
Date: Tue, 27 Aug 2024 16:00:29 +0100
Subject: wip: update TestLambdaHandler & lambda_handler function

---
 src/load_lambda.py        | 19 +++++++++++--------
 tests/test_load_lambda.py | 12 +++++++++---
 2 files changed, 20 insertions(+), 11 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 11d1d70..39fa27d 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -23,18 +23,21 @@ logging.getLogger("botocore").setLevel(logging.INFO)
 def lambda_handler(event, context):
     try:
         uploaded_tables = upload_dfs_to_database()
-        if not uploaded_tables["uploaded"]:
+        if uploaded_tables["not_uploaded"]:
             return {
                 "statusCode": 200,
                 "body": json.dumps("No dataframes were uploaded."),
             }
-        return {
-            "statusCode": 200,
-            "body": json.dumps(
-                f"""The following dataframes were uploaded successfully: 
-                {uploaded_tables["uploaded"]} ."""
-            ),
-        }
+
+        if uploaded_tables["uploaded"]:
+            return {
+                "statusCode": 200,
+                "body": json.dumps(
+                    f"""The following dataframes were uploaded successfully: 
+                    {uploaded_tables["uploaded"]} ."""
+                ),
+            }
+
     except Exception as e:
         logger.error(f"Error: {e}", exc_info=True)
         return {"statusCode": 500, "body": json.dumps("Internal server error.")}
diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py
index a29b75a..9286e48 100644
--- a/tests/test_load_lambda.py
+++ b/tests/test_load_lambda.py
@@ -35,7 +35,7 @@ class TestLambdaHandler:
     def test_lambda_handler_returns_success(self, mocker):
         mocker.patch(
             "src.load_lambda.upload_dfs_to_database",
-            return_value={"uploaded": ["table_one", "table_two"]},
+            return_value={"uploaded": ["table_one", "table_two"], "not_uploaded": []},
         )
         result = lambda_handler(None, None)
         assert result["statusCode"] == 200
@@ -45,14 +45,20 @@ class TestLambdaHandler:
     def test_lambda_handler_does_not_upload_anything(self, mocker):
         mocker.patch(
             "src.load_lambda.upload_dfs_to_database",
-            return_value={"uploaded": []},
+            return_value={"uploaded": [], "not_uploaded": []},
         )
         result = lambda_handler(None, None)
         assert result["statusCode"] == 200
         assert "No dataframes were uploaded" in result["body"]
 
     def test_lambda_handler_returns_exception(self, mocker):
-        pass
+        mocker.patch(
+            "src.load_lambda.upload_dfs_to_database",
+            return_value={"test": []},
+        )
+
+        with pytest.raises(Exception):
+            lambda_handler(None, None)
 
 
 class TestRetrieveSecrets:
-- 
cgit v1.2.3


From 27f89b78775f9b6fd8d3d560689c53db2beb1b64 Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Tue, 27 Aug 2024 16:39:38 +0100
Subject: add logger error to lambda handler

---
 src/load_lambda.py | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 39fa27d..9e15af3 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -5,6 +5,7 @@ import pyarrow.parquet as pq
 from io import BytesIO
 import logging
 import json
+import traceback
 from sqlalchemy import create_engine
 
 
@@ -28,8 +29,7 @@ def lambda_handler(event, context):
                 "statusCode": 200,
                 "body": json.dumps("No dataframes were uploaded."),
             }
-
-        if uploaded_tables["uploaded"]:
+        elif uploaded_tables["uploaded"]:
             return {
                 "statusCode": 200,
                 "body": json.dumps(
@@ -37,10 +37,12 @@ def lambda_handler(event, context):
                     {uploaded_tables["uploaded"]} ."""
                 ),
             }
-
+        else:
+            logger.error(f"error")
+            return {"error"}
     except Exception as e:
-        logger.error(f"Error: {e}", exc_info=True)
-        return {"statusCode": 500, "body": json.dumps("Internal server error.")}
+        logger.error({e})
+        return {"statusCode": 500, "body": {e}}
 
 
 def retrieve_secrets(client=None, secret_name=None):
-- 
cgit v1.2.3


From 0915d4fe4e151d6b593467129b51a1322398fc04 Mon Sep 17 00:00:00 2001
From: Ellie <ecsymonds@gmail.com>
Date: Tue, 27 Aug 2024 17:27:21 +0100
Subject: add json.loads

---
 src/load_lambda.py | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 9e15af3..7339ab9 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -64,7 +64,7 @@ def retrieve_secrets(client=None, secret_name=None):
         logger.error(f"Secret {secret_name} does not contain a SecretString")
         raise ValueError(f"Secret {secret_name} does not contain a SecretString")
 
-    return json.loads(get_secret_value_response["SecretString"])
+    return get_secret_value_response["SecretString"]
 
 
 # connect to database, slightly different way of doing it, to allow manipulation through pandas
@@ -72,10 +72,10 @@ def retrieve_secrets(client=None, secret_name=None):
 
 def connect_to_db_and_return_engine(sm_secret=None):
     if sm_secret is None:
-        sm_secret = retrieve_secrets()
+        sm_secret = json.loads(retrieve_secrets())
 
     try:
-        secrets = json.loads(sm_secret)
+        secrets = sm_secret
         host = secrets["host"]
         port = secrets["port"]
         user = secrets["user"]
@@ -171,13 +171,14 @@ def upload_dfs_to_database():
     ]
 
     for file_name, df in dict_of_dfs.items():
+        print(df)
         if file_name in immutable_df_dict:
             table_name = file_name.split(".")[0]
+            print(table_name, "<<<<<")
             try:
                 df.to_sql(
                     table_name,
                     con=db_engine,
-                    schema="project_team_2",
                     if_exists="append",
                     index=False,
                 )
-- 
cgit v1.2.3


From 03a5959df25f74d52ed5393c2a5af6b1b9eb34c9 Mon Sep 17 00:00:00 2001
From: T-Aji <tolujbd2@gmail.com>
Date: Wed, 28 Aug 2024 12:48:13 +0100
Subject: refactored functs to include columns instead of drop columns

---
 src/load_lambda.py                       |   5 +-
 src/transform_lambda/dataframes.py       | 157 ++++++++++++++++++++-----------
 src/transform_lambda/transform_lambda.py |   5 +-
 tests/test_dataframes.py                 |   2 +-
 4 files changed, 111 insertions(+), 58 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 7339ab9..926b4db 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -134,6 +134,9 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
                     file_obj = client.get_object(Bucket=bucket_name, Key=file_key)
                     parquet_file = pq.ParquetFile(BytesIO(file_obj["Body"].read()))
                     df = parquet_file.read().to_pandas()
+                    print("df", df)
+                    print("type", type(df))
+                    print(df.columns)
                     dfs[file_key] = df
                 except ClientError as e:
                     logger.error(f"Unable to retrieve S3 object {file_key}: {e}")
@@ -148,7 +151,7 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
     except ClientError as client_error:
         logger.error(f"Unable to list objects: {client_error}")
         raise
-
+    print()
     return dfs
 
 
diff --git a/src/transform_lambda/dataframes.py b/src/transform_lambda/dataframes.py
index 2a46bd6..bf0556b 100644
--- a/src/transform_lambda/dataframes.py
+++ b/src/transform_lambda/dataframes.py
@@ -37,30 +37,28 @@ def create_fact_sales_order(dict_of_df):
     df_sales["agreed_payment_date"] = pd.to_datetime(
         df_sales["agreed_payment_date"], format="%Y-%m-%d"
     )
-    df_sales = df_sales.drop(labels=["created_at", "last_updated"], axis=1)
-
-    df_sales.reset_index(inplace=True)
-    return df_sales
+    fact_sales = df_sales.loc[:,
+        [
+            "sales_order_id",
+            "created_date",
+            "created_time",
+            "last_updated_date",
+            "last_updated_time",
+            "staff_id",
+            "counterparty_id",
+            "units_sold",
+            "unit_price",
+            "currency_id",
+            "design_id",
+            "agreed_payment_date",
+            "agreed_delivery_date",
+            "agreed_delivery_location_id"
+        ],
+    ]
+    fact_sales.rename(columns={"staff_id": "sales_staff_id"}).reset_index(inplace=True)
+    
 
-    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["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["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 = df_sales.drop(labels=["created_at", "last_updated"], axis=1)
-    df_sales.reset_index(inplace=True)
-    return df_sales
+    return fact_sales
 
 
 # no test, same as fact_payment
@@ -83,9 +81,27 @@ def create_fact_purchase_orders(dict_of_df):
     df_po["agreed_payment_date"] = pd.to_datetime(
         df_po["agreed_payment_date"], format="%Y-%m-%d"
     )
-    df_po = df_po.drop(labels=["created_at", "last_updated"], axis=1)
-    df_po.reset_index(inplace=True)
-    return df_po
+    fact_purchase_order = df_po.loc[:,
+        [                           
+        "purchase_order_id",
+        "created_date",
+        "created_time",
+        "last_updated_date",
+        "last_updated_time",
+        "staff_id",
+        "counterparty_id",
+        "item_code",
+        "item_quantity",
+        "item_unit_price",
+        "currency_id",
+        "agreed_delivery_date",
+        "agreed_payment_date",
+        "agreed_delivery_location_id"
+        ]
+
+    ]
+    fact_purchase_order.reset_index(inplace=True)
+    return fact_purchase_order
 
 
 # test passed
@@ -109,38 +125,57 @@ def create_fact_payment(dict_of_df):
     df_payment["payment_date"] = pd.to_datetime(
         df_payment["payment_date"], format="%Y-%m-%d"
     )
-    df_payment = df_payment.drop(labels=["created_at", "last_updated"], axis=1)
-
-    df_payment.reset_index(inplace=True)
-    return df_payment
+    fact_payment = df_payment.loc[:,
+        [
+        "payment_id",
+        "created_date",
+        "created_time",
+        "last_updated_date",
+        "last_updated_time",
+        "transaction_id",
+        "counterparty_id",
+        "payment_amount",
+        "currency_id",
+        "payment_type_id",
+        "paid",
+        "payment_date"
+        ]
+    ]
+    fact_payment.reset_index(inplace=True)
+    return fact_payment
 
 
 # test passed
 
 
 def create_dim_transaction(dict_of_df):
-    df_transaction = dict_of_df["transaction"].drop(
-        labels=["created_at", "last_updated"], axis=1
-    )
-    return df_transaction
+    dim_transaction = dict_of_df["transaction"].loc[:,
+        [
+        "transaction_id",
+        "transaction_type",
+        "sales_order_id",
+        "purchase_order_id"
+        ]
+    ]
+    return dim_transaction
 
 
 # test passed
 
 
 def create_dim_location(dict_of_df):
-    df_loc = (
-        dict_of_df["address"]
-        .drop(labels=["created_at", "last_updated"], axis=1)
+    dim_location = (
+        dict_of_df["address"].drop(labels=["created_at", "last_updated"], axis=1)
         .rename(columns={"address_id": "location_id"})
     )
-    return df_loc
+    return dim_location
 
 
 def create_dim_counterparty(dict_of_df):
     df_prefixed_address = (
         dict_of_df["address"]
         .drop(labels=["created_at", "last_updated"], axis=1)
+        .rename(columns={"phone": "phone_number"})
         .add_prefix("counterparty_legal_", axis=1)
     )
     df_cp = pd.merge(
@@ -149,15 +184,18 @@ def create_dim_counterparty(dict_of_df):
         left_on="legal_address_id",
         right_on="counterparty_legal_address_id",
         how="inner",
-    )
-    df_cp.drop(
-        columns=[
+    )#.dropna(inplace=True)
+    dim_counterparty = df_cp.drop(
+        labels=[
             "legal_address_id",
             "counterparty_legal_address_id",
-        ],
-        inplace=True,
+            "created_at",
+            "last_updated",
+            "commercial_contact",
+            "delivery_contact"
+        ], axis=1
     )
-    return df_cp
+    return dim_counterparty
 
 
 # test passed
@@ -179,6 +217,7 @@ def create_dim_date(dict_of_df):
     sr_date = pd.array(pd.concat(list_of_date_columns), dtype="datetime64[ns]")
     df_date = pd.DataFrame(data=sr_date, columns=["date_id"])
     df_date.drop_duplicates(inplace=True)
+    # df_date.dropna(inplace=True)
     df_date["year"] = df_date["date_id"].dt.year
     df_date["month"] = df_date["date_id"].dt.month
     df_date["day"] = df_date["date_id"].dt.day
@@ -210,10 +249,11 @@ def scrape_currency_names():
 
 def create_dim_currency(dict_of_df, names=scrape_currency_names()):
     df_cur = dict_of_df["currency"].drop(labels=["created_at", "last_updated"], axis=1)
-    dim_cur = pd.merge(
-        df_cur, names, left_on="currency_code", right_on="currency_code", how="inner"
+    dim_currency = pd.merge(
+        df_cur, names, left_on="currency_code", right_on="currency_code", how="left"
     )
-    return dim_cur
+    dim_currency.drop_duplicates(inplace=True)
+    return dim_currency
 
 
 # tests passed
@@ -221,7 +261,12 @@ def create_dim_currency(dict_of_df, names=scrape_currency_names()):
 
 def create_dim_payment_type(dict_of_df):
     df_payment_type = dict_of_df["payment_type"]
-    dim_payment_type = df_payment_type.loc[:, ["payment_type_id", "payment_type_name"]]
+    dim_payment_type = df_payment_type.loc[:, 
+        [
+            "payment_type_id", 
+            "payment_type_name"
+        ]
+    ]
     return dim_payment_type
 
 
@@ -230,8 +275,13 @@ def create_dim_payment_type(dict_of_df):
 
 def create_dim_design(dict_of_df):
     df_design = dict_of_df["design"]
-    dim_design = df_design.loc[
-        :, ["design_id", "design_name", "file_name", "file_location"]
+    dim_design = df_design.loc[:, 
+        [
+            "design_id", 
+            "design_name", 
+            "file_name", 
+            "file_location"
+        ]
     ]
     return dim_design
 
@@ -243,15 +293,14 @@ def create_dim_staff(dict_of_df):
     staff_department = pd.merge(
         dict_of_df["staff"], dict_of_df["department"], on="department_id", how="left"
     )
-    dim_staff = staff_department.loc[
-        :,
+    dim_staff = staff_department.loc[:,
         [
             "staff_id",
             "first_name",
             "last_name",
             "department_name",
             "location",
-            "email_address",
-        ],
+            "email_address"
+        ]
     ]
     return dim_staff
diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py
index 93b2284..1453c6c 100644
--- a/src/transform_lambda/transform_lambda.py
+++ b/src/transform_lambda/transform_lambda.py
@@ -42,7 +42,7 @@ TABLES = [
     "department",
     "currency",
     "design",
-    "payment_type",
+    "payment_type"
 ]
 
 
@@ -73,7 +73,8 @@ def lambda_handler(event, context):
             "fact_payment": create_fact_payment(dict_of_df),
             "dim_currency": create_dim_currency(dict_of_df),
         }
-
+        print(immutable_df_dict.values())
+        print(mutable_df_dict.values())
         status = process_to_parquet_and_upload_to_s3(
             existing_s3_files, immutable_df_dict, mutable_df_dict, bucket
         )
diff --git a/tests/test_dataframes.py b/tests/test_dataframes.py
index ea7bad1..7dd592a 100644
--- a/tests/test_dataframes.py
+++ b/tests/test_dataframes.py
@@ -1,4 +1,4 @@
-from src.dataframes import *
+from src.transform_lambda.dataframes import *
 import pandas as pd
 from unittest.mock import patch
 from datetime import datetime as dt
-- 
cgit v1.2.3


From d064b2ec2c7393f8de50560a7edfe100851bfea3 Mon Sep 17 00:00:00 2001
From: T-Aji <tolujbd2@gmail.com>
Date: Wed, 28 Aug 2024 14:39:13 +0100
Subject: debugging load_lambda

---
 src/load_lambda.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 926b4db..272cb8c 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -164,13 +164,13 @@ def upload_dfs_to_database():
         "dim_date.parquet",  # this needs to be mutable
         "dim_location.parquet",
         "dim_staff.parquet",
-        "dim_design.parquet",
+        "dim_design.parquet"
     ]
     mutable_df_dict = [
         "fact_sales_order",
         "fact_purchase_order",
         "fact_payment",
-        "dim_currency",
+        "dim_currency"
     ]
 
     for file_name, df in dict_of_dfs.items():
@@ -182,6 +182,7 @@ def upload_dfs_to_database():
                 df.to_sql(
                     table_name,
                     con=db_engine,
+                    schema="project_team_2",
                     if_exists="append",
                     index=False,
                 )
-- 
cgit v1.2.3


From 6235a2bb04b60d57a41196b07bbf0296920c6980 Mon Sep 17 00:00:00 2001
From: T-Aji <tolujbd2@gmail.com>
Date: Wed, 28 Aug 2024 17:52:45 +0100
Subject: wip commit

---
 src/load_lambda.py                       | 174 +++++++++++++++++++------------
 src/transform_lambda/dataframes.py       |   8 +-
 src/transform_lambda/transform_lambda.py |   2 +-
 tests/test_transform_lambda.py           |   2 +-
 4 files changed, 115 insertions(+), 71 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 272cb8c..cdcf105 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -7,7 +7,8 @@ import logging
 import json
 import traceback
 from sqlalchemy import create_engine
-
+from datetime import datetime as dt
+import re
 
 logger = logging.getLogger(__name__)
 
@@ -15,10 +16,10 @@ logging.basicConfig(
     format="{asctime} - {levelname} - {message}",
     style="{",
     datefmt="%Y-%m-%d %H:%M",
-    level=logging.DEBUG,
+    level=logging.INFO,
 )
-
-logging.getLogger("botocore").setLevel(logging.INFO)
+# logging.getLogger("botocore").setLevel(logging.INFO)
+# logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
 
 
 def lambda_handler(event, context):
@@ -38,10 +39,10 @@ def lambda_handler(event, context):
                 ),
             }
         else:
-            logger.error(f"error")
+            logger.error(f"error", exc_info=True)
             return {"error"}
     except Exception as e:
-        logger.error({e})
+        logger.error({e}, exc_info=True)
         return {"statusCode": 500, "body": {e}}
 
 
@@ -58,10 +59,10 @@ def retrieve_secrets(client=None, secret_name=None):
         get_secret_value_response = client.get_secret_value(SecretId=secret_name)
         print(get_secret_value_response)
     except ClientError as e:
-        logger.error(f"Failed to retrieve secret {secret_name}: {str(e)}")
+        logger.error(f"Failed to retrieve secret {secret_name}: {str(e)}", exc_info=True)
         raise e
     except KeyError:
-        logger.error(f"Secret {secret_name} does not contain a SecretString")
+        logger.error(f"Secret {secret_name} does not contain a SecretString", exc_info=True)
         raise ValueError(f"Secret {secret_name} does not contain a SecretString")
 
     return get_secret_value_response["SecretString"]
@@ -86,7 +87,7 @@ def connect_to_db_and_return_engine(sm_secret=None):
         engine = create_engine(conn_str)
         return engine
     except Exception as e:
-        logger.error(f"Interface error: {e}")
+        logger.error(f"Interface error: {e}", exc_info=True)
         raise RuntimeError("Failed to create database engine")
 
 
@@ -97,7 +98,7 @@ def get_transform_bucket(client=None):
     try:
         response = client.list_buckets()
     except ClientError as e:
-        logger.error(f"Error listing S3 buckets: {e}")
+        logger.error(f"Error listing S3 buckets: {e}", exc_info=True)
         raise RuntimeError("Error listing S3 buckets")
 
     transform_bucket_filter = [
@@ -107,7 +108,7 @@ def get_transform_bucket(client=None):
     ]
 
     if not transform_bucket_filter:
-        logger.error("No transform bucket found")
+        logger.error("No transform bucket found", exc_info=True)
         raise ValueError("No transform bucket found")
 
     return transform_bucket_filter[0]
@@ -117,41 +118,78 @@ def get_transform_bucket(client=None):
 # convert parquet files into dataframes
 # return a dictionary of dataframes with name as key, and dataframe object as value
 
+def get_latest_timestamp(existing_files):
+    if existing_files:
+        all_datetimes = []
+        for file_name in existing_files:
+            match = re.search(r"\/(.+/).+_(.+)\.parquet", file_name)
+            if match:
+                datetime_str = "".join(match.group(1, 2))
+                all_datetimes.append(
+                    dt.strptime(datetime_str, "%Y/%m/%d/%H:%M:%S")
+                )
+        return max(all_datetimes) if all_datetimes else dt.min
+    return existing_files
 
 def convert_parquet_files_to_dfs(bucket_name=None, client=None):
+    mutable_df_dict = [
+        "dim_currency",
+        "fact_sales_order",
+        "fact_purchase_order",
+        "fact_payment"
+        
+    ]
+
     try:
         if client is None:
             client = boto3.client("s3")
         if bucket_name is None:
             bucket_name = get_transform_bucket()
         files = client.list_objects_v2(Bucket=bucket_name)
-
+        
         dfs = {}
         if "Contents" in files:
-            for file in files["Contents"]:
-                file_key = file["Key"]
+            s3_key_list = [file["Key"]for file in files["Contents"]]
+            immutables_l = []
+            mutables_d = {prefix:[] for prefix in mutable_df_dict}
+            for tab, s3_key in mutables_d.items():
+                for file in s3_key_list:
+                    if tab in file:
+                        s3_key.append(file)
+                    elif "2024" not in file:
+                        immutables_l.append(file)
+                    else:
+                        continue
+            immutables_l = list(set(immutables_l))
+            print(mutables_d,'mutables_d')
+            latest_s3_keys = []
+            for k,v in mutables_d.items():
+	            latest_s3_keys.append(dt.strftime(get_latest_timestamp(v), f"{k}/%Y/%m/%d/{k}_%H:%M:%S.parquet"))
+            print(latest_s3_keys,'latest')
+            print(immutables_l,'immutables_l')
+            for file_key in latest_s3_keys+immutables_l:
                 try:
                     file_obj = client.get_object(Bucket=bucket_name, Key=file_key)
                     parquet_file = pq.ParquetFile(BytesIO(file_obj["Body"].read()))
                     df = parquet_file.read().to_pandas()
-                    print("df", df)
-                    print("type", type(df))
-                    print(df.columns)
-                    dfs[file_key] = df
+                    df_without_nulls = df.dropna()
+                    #print("df_without_nulls", df_without_nulls)
+                    #print("type", type(df_without_nulls))
+                    #print(df_without_nulls.columns)
+                    dfs[file_key] = df_without_nulls
                 except ClientError as e:
-                    logger.error(f"Unable to retrieve S3 object {file_key}: {e}")
+                    logger.error(f"Unable to retrieve S3 object {file_key}: {e}", exc_info=True)
                 except Exception as e:
-                    logger.error(f"Unable to process file {file_key}: {e}")
+                    logger.error(f"Unable to process file {file_key}: {e}", exc_info=True)
         else:
-            logger.error(f"No files found in {bucket_name}.")
+            logger.error(f"No files found in {bucket_name}.", exc_info=True)
             return {}
     except ValueError as value_error:
-        logger.error(f"Unable to list objects: {value_error}")
+        logger.error(f"Unable to list objects: {value_error}", exc_info=True)
         raise
     except ClientError as client_error:
-        logger.error(f"Unable to list objects: {client_error}")
+        logger.error(f"Unable to list objects: {client_error}", exc_info=True)
         raise
-    print()
     return dfs
 
 
@@ -160,53 +198,57 @@ def upload_dfs_to_database():
     dict_of_dfs = convert_parquet_files_to_dfs()
     db_engine = connect_to_db_and_return_engine()
     immutable_df_dict = [
-        "dim_counterparty.parquet",
-        "dim_date.parquet",  # this needs to be mutable
-        "dim_location.parquet",
-        "dim_staff.parquet",
-        "dim_design.parquet"
+        # #"dim_counterparty.parquet",
+        # "dim_date.parquet",  # this needs to be mutable
+        # "dim_location.parquet",
+        # "dim_staff.parquet",
+        # "dim_design.parquet"
     ]
     mutable_df_dict = [
+        "dim_currency",
         "fact_sales_order",
         "fact_purchase_order",
-        "fact_payment",
-        "dim_currency"
+        "fact_payment"
+        
     ]
-
-    for file_name, df in dict_of_dfs.items():
-        print(df)
-        if file_name in immutable_df_dict:
-            table_name = file_name.split(".")[0]
-            print(table_name, "<<<<<")
-            try:
-                df.to_sql(
-                    table_name,
-                    con=db_engine,
-                    schema="project_team_2",
-                    if_exists="append",
-                    index=False,
-                )
-                upload_status["uploaded"].append(table_name)
-            except Exception as e:
-                logger.error(f"Error uploading dataframe {file_name} to database: {e}")
-                raise
-        elif file_name.rsplit("_", 1)[0] in mutable_df_dict:
-            table_name = file_name.rsplit("_", 1)[0]
-            try:
-                df.to_sql(
-                    table_name,
-                    con=db_engine,
-                    schema="project_team_2",
-                    if_exists="append",
-                    index=False,
-                )
-                upload_status["uploaded"].append(table_name)
-            except Exception as e:
-                logger.error(f"Error uploading dataframe {file_name} to database: {e}")
-                raise
-        else:
-            upload_status["not_uploaded"].append(file_name)
-            logger.error(f"{file_name} does not correspond with table in database")
+    with db_engine.begin() as connection:
+        for file_name, df in dict_of_dfs.items():
+            print(df.dtypes, "dtypes")
+            print(df.head())
+            if file_name in immutable_df_dict:
+                table_name = file_name.split(".")[0]
+                print(table_name, "<<<<<")
+                try:
+                    df.to_sql(
+                        table_name,
+                        con=connection,
+                        schema="project_team_2",
+                        if_exists="append",
+                        index=False,
+                    )
+                    upload_status["uploaded"].append(table_name)
+                    print(upload_status)
+                except Exception as e:
+                    logger.error(f"Error uploading dataframe {file_name} to database: {e}", exc_info=True)
+                    raise
+            elif file_name.split("/")[0] in mutable_df_dict:
+                table_name = file_name.split("/")[0]
+                print(table_name, "<<<<<<<TABLE NAME")
+                try:
+                    df.to_sql(
+                        table_name,
+                        con=connection,
+                        schema="project_team_2",
+                        if_exists="append",
+                        index=False,
+                    )
+                    upload_status["uploaded"].append(table_name)
+                except Exception as e:
+                    logger.error(f"Error uploading dataframe {file_name} to database: {e}", exc_info=True)
+                    raise
+            else:
+                upload_status["not_uploaded"].append(file_name)
+            logger.error(f"{file_name} does not correspond with table in database", exc_info=True)
     db_engine.dispose()
     return upload_status
 
diff --git a/src/transform_lambda/dataframes.py b/src/transform_lambda/dataframes.py
index bf0556b..e89a6b2 100644
--- a/src/transform_lambda/dataframes.py
+++ b/src/transform_lambda/dataframes.py
@@ -18,7 +18,7 @@ import requests
 
 # no test, same as fact_payment
 def create_fact_sales_order(dict_of_df):
-    df_sales = dict_of_df["sales_order"]
+    df_sales = dict_of_df["sales_order"].rename(columns={"staff_id": "sales_staff_id"})
     df_sales.index.name = "sales_record_id"
 
     df_sales["created_date"] = df_sales["created_at"].astype("datetime64[ns]").dt.date
@@ -44,7 +44,7 @@ def create_fact_sales_order(dict_of_df):
             "created_time",
             "last_updated_date",
             "last_updated_time",
-            "staff_id",
+            "sales_staff_id",
             "counterparty_id",
             "units_sold",
             "unit_price",
@@ -55,7 +55,7 @@ def create_fact_sales_order(dict_of_df):
             "agreed_delivery_location_id"
         ],
     ]
-    fact_sales.rename(columns={"staff_id": "sales_staff_id"}).reset_index(inplace=True)
+    fact_sales.reset_index(inplace=True)
     
 
     return fact_sales
@@ -253,6 +253,8 @@ def create_dim_currency(dict_of_df, names=scrape_currency_names()):
         df_cur, names, left_on="currency_code", right_on="currency_code", how="left"
     )
     dim_currency.drop_duplicates(inplace=True)
+    dim_currency.astype({"currency_name": "string", "currency_code": "string"})
+    print(dim_currency.dtypes, "<<<<<<<<<Dtype")
     return dim_currency
 
 
diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py
index 1453c6c..0b5748b 100644
--- a/src/transform_lambda/transform_lambda.py
+++ b/src/transform_lambda/transform_lambda.py
@@ -5,7 +5,7 @@ import logging
 import pandas as pd
 import pyarrow as pa
 import pyarrow.parquet as pq
-from dataframes import *
+from src.transform_lambda.dataframes import *
 from botocore.exceptions import ClientError
 from pg8000.native import Connection, InterfaceError
 from datetime import datetime
diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py
index 5ed743e..308dc65 100644
--- a/tests/test_transform_lambda.py
+++ b/tests/test_transform_lambda.py
@@ -1,4 +1,4 @@
-from src.transform_lambda import (
+from src.transform_lambda.transform_lambda import (
     read_from_s3_subfolder_to_df,
     list_existing_s3_files,
     bucket_name,
-- 
cgit v1.2.3


From d396cd95d660fb76188ef887fc268d20aeeb5352 Mon Sep 17 00:00:00 2001
From: Ang Bel <anzelikabelotelova@Anzelikas-MacBook-Air.local>
Date: Wed, 28 Aug 2024 22:46:00 +0100
Subject: fix: adds missing dataframes and resolves tables upload to end data
 warehouse in case the table is empty

---
 .gitignore                               |  6 +++++-
 src/load_lambda.py                       | 24 +++++++++++++-----------
 src/transform_lambda/dataframes.py       | 19 ++++++++++++++-----
 src/transform_lambda/transform_lambda.py |  4 +++-
 4 files changed, 35 insertions(+), 18 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/.gitignore b/.gitignore
index 6aa03fc..480ae4b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -14,4 +14,8 @@ __pycache__/
 
 # OS-Related Files
 .DS_Store
-venv
\ No newline at end of file
+venv
+
+#files
+/dim_*
+/fact_*
\ No newline at end of file
diff --git a/src/load_lambda.py b/src/load_lambda.py
index cdcf105..8f921b8 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -161,18 +161,15 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
                     else:
                         continue
             immutables_l = list(set(immutables_l))
-            print(mutables_d,'mutables_d')
             latest_s3_keys = []
             for k,v in mutables_d.items():
 	            latest_s3_keys.append(dt.strftime(get_latest_timestamp(v), f"{k}/%Y/%m/%d/{k}_%H:%M:%S.parquet"))
-            print(latest_s3_keys,'latest')
-            print(immutables_l,'immutables_l')
-            for file_key in latest_s3_keys+immutables_l:
+            for file_key in immutables_l+latest_s3_keys:
                 try:
                     file_obj = client.get_object(Bucket=bucket_name, Key=file_key)
                     parquet_file = pq.ParquetFile(BytesIO(file_obj["Body"].read()))
                     df = parquet_file.read().to_pandas()
-                    df_without_nulls = df.dropna()
+                    df_without_nulls = df.dropna(how='all') #>> can't do 'any' (default) because we lose rows in dim_location
                     #print("df_without_nulls", df_without_nulls)
                     #print("type", type(df_without_nulls))
                     #print(df_without_nulls.columns)
@@ -202,12 +199,14 @@ def upload_dfs_to_database():
         # "dim_date.parquet",  # this needs to be mutable
         # "dim_location.parquet",
         # "dim_staff.parquet",
-        # "dim_design.parquet"
+        # "dim_design.parquet",
+        # 'dim_transaction.parquet' #This one was missing,
+        'dim_payment_type.parquet'
     ]
     mutable_df_dict = [
-        "dim_currency",
-        "fact_sales_order",
-        "fact_purchase_order",
+        # "dim_currency",
+        # "fact_sales_order",
+        # "fact_purchase_order",
         "fact_payment"
         
     ]
@@ -215,7 +214,9 @@ def upload_dfs_to_database():
         for file_name, df in dict_of_dfs.items():
             print(df.dtypes, "dtypes")
             print(df.head())
-            if file_name in immutable_df_dict:
+            print(file_name,"<<< FILE NAME")
+            print(immutable_df_dict,"<<<IMMUTABLE_DF_DICT")
+            if  file_name in immutable_df_dict: 
                 table_name = file_name.split(".")[0]
                 print(table_name, "<<<<<")
                 try:
@@ -248,7 +249,8 @@ def upload_dfs_to_database():
                     raise
             else:
                 upload_status["not_uploaded"].append(file_name)
-            logger.error(f"{file_name} does not correspond with table in database", exc_info=True)
+                logger.error(f"{file_name} does not correspond with table in database", exc_info=True)
+            print(upload_status)
     db_engine.dispose()
     return upload_status
 
diff --git a/src/transform_lambda/dataframes.py b/src/transform_lambda/dataframes.py
index e89a6b2..c823b87 100644
--- a/src/transform_lambda/dataframes.py
+++ b/src/transform_lambda/dataframes.py
@@ -19,7 +19,6 @@ import requests
 # no test, same as fact_payment
 def create_fact_sales_order(dict_of_df):
     df_sales = dict_of_df["sales_order"].rename(columns={"staff_id": "sales_staff_id"})
-    df_sales.index.name = "sales_record_id"
 
     df_sales["created_date"] = df_sales["created_at"].astype("datetime64[ns]").dt.date
     df_sales["created_time"] = (
@@ -55,9 +54,11 @@ def create_fact_sales_order(dict_of_df):
             "agreed_delivery_location_id"
         ],
     ]
+    fact_sales.convert_dtypes()
+    fact_sales.index = pd.RangeIndex(1, len(fact_sales.index) + 1)
+    fact_sales.index.name = "sales_record_id"
     fact_sales.reset_index(inplace=True)
-    
-
+    fact_sales.dropna(inplace=True)
     return fact_sales
 
 
@@ -66,7 +67,6 @@ 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"] = 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
@@ -100,7 +100,11 @@ def create_fact_purchase_orders(dict_of_df):
         ]
 
     ]
+    fact_purchase_order.convert_dtypes()
+    fact_purchase_order.index = pd.RangeIndex(1, len(fact_purchase_order.index) + 1)
+    fact_purchase_order.index.name = "purchase_record_id"
     fact_purchase_order.reset_index(inplace=True)
+    fact_purchase_order.dropna(inplace=True)
     return fact_purchase_order
 
 
@@ -109,7 +113,6 @@ 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"] = (
         df_payment["created_at"].astype("datetime64[ns]").dt.date
     )
@@ -141,7 +144,12 @@ def create_fact_payment(dict_of_df):
         "payment_date"
         ]
     ]
+    fact_payment.convert_dtypes()
+    fact_payment.index = pd.RangeIndex(1, len(fact_payment.index) + 1)
+    fact_payment.index.name = "payment_record_id"
     fact_payment.reset_index(inplace=True)
+    fact_payment.dropna(inplace=True)
+    fact_payment = fact_payment.astype({'currency_id':'int','payment_id':'int'})
     return fact_payment
 
 
@@ -157,6 +165,7 @@ def create_dim_transaction(dict_of_df):
         "purchase_order_id"
         ]
     ]
+    #dim_transaction = dim_transaction.astype({"sales_order_id":"Int64","purchase_order_id":"Int64"})
     return dim_transaction
 
 
diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py
index 0b5748b..5ea8cf0 100644
--- a/src/transform_lambda/transform_lambda.py
+++ b/src/transform_lambda/transform_lambda.py
@@ -65,13 +65,15 @@ def lambda_handler(event, context):
             "dim_location": create_dim_location(dict_of_df),
             "dim_staff": create_dim_staff(dict_of_df),
             "dim_design": create_dim_design(dict_of_df),
+            "dim_transaction": create_dim_transaction(dict_of_df),
+            "dim_payment_type": create_dim_payment_type(dict_of_df)
         }
 
         mutable_df_dict = {
             "fact_sales_order": create_fact_sales_order(dict_of_df),
             "fact_purchase_order": create_fact_purchase_orders(dict_of_df),
             "fact_payment": create_fact_payment(dict_of_df),
-            "dim_currency": create_dim_currency(dict_of_df),
+            "dim_currency": create_dim_currency(dict_of_df)
         }
         print(immutable_df_dict.values())
         print(mutable_df_dict.values())
-- 
cgit v1.2.3


From 48e7daec8b5435a696fe572fd51dcbc8f9604a2d Mon Sep 17 00:00:00 2001
From: Ang Bel <anzelikabelotelova@Anzelikas-MacBook-Air.local>
Date: Thu, 29 Aug 2024 09:47:58 +0100
Subject: fix: added comma. Code complete and uploads all tables in one go if
 no data exists per each table

---
 src/load_lambda.py | 19 +++++++++----------
 1 file changed, 9 insertions(+), 10 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 8f921b8..941ae97 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -57,7 +57,6 @@ def retrieve_secrets(client=None, secret_name=None):
 
     try:
         get_secret_value_response = client.get_secret_value(SecretId=secret_name)
-        print(get_secret_value_response)
     except ClientError as e:
         logger.error(f"Failed to retrieve secret {secret_name}: {str(e)}", exc_info=True)
         raise e
@@ -195,18 +194,18 @@ def upload_dfs_to_database():
     dict_of_dfs = convert_parquet_files_to_dfs()
     db_engine = connect_to_db_and_return_engine()
     immutable_df_dict = [
-        # #"dim_counterparty.parquet",
-        # "dim_date.parquet",  # this needs to be mutable
-        # "dim_location.parquet",
-        # "dim_staff.parquet",
-        # "dim_design.parquet",
-        # 'dim_transaction.parquet' #This one was missing,
+        "dim_counterparty.parquet",
+        "dim_date.parquet",  # this needs to be mutable
+        "dim_location.parquet",
+        "dim_staff.parquet",
+        "dim_design.parquet",
+        'dim_transaction.parquet', #This one was missing,
         'dim_payment_type.parquet'
     ]
     mutable_df_dict = [
-        # "dim_currency",
-        # "fact_sales_order",
-        # "fact_purchase_order",
+        "dim_currency",
+        "fact_sales_order",
+        "fact_purchase_order",
         "fact_payment"
         
     ]
-- 
cgit v1.2.3


From 42ad135b25044bb1c7ab8a553f038c8da9de0f75 Mon Sep 17 00:00:00 2001
From: "deepsource-autofix[bot]"
 <62050782+deepsource-autofix[bot]@users.noreply.github.com>
Date: Thu, 29 Aug 2024 08:57:48 +0000
Subject: style: format code with Autopep8, Black and Ruff Formatter

This commit fixes the style issues introduced in 48e7dae according to the output
from Autopep8, Black and Ruff Formatter.

Details: https://github.com/ajschofield/de-project-bentley/pull/107
---
 src/load_lambda.py                       |  78 +++++++++++++--------
 src/transform_lambda/dataframes.py       | 116 ++++++++++++++-----------------
 src/transform_lambda/transform_lambda.py |   6 +-
 3 files changed, 105 insertions(+), 95 deletions(-)

(limited to 'src/load_lambda.py')

diff --git a/src/load_lambda.py b/src/load_lambda.py
index 941ae97..86189dc 100644
--- a/src/load_lambda.py
+++ b/src/load_lambda.py
@@ -58,10 +58,14 @@ def retrieve_secrets(client=None, secret_name=None):
     try:
         get_secret_value_response = client.get_secret_value(SecretId=secret_name)
     except ClientError as e:
-        logger.error(f"Failed to retrieve secret {secret_name}: {str(e)}", exc_info=True)
+        logger.error(
+            f"Failed to retrieve secret {secret_name}: {str(e)}", exc_info=True
+        )
         raise e
     except KeyError:
-        logger.error(f"Secret {secret_name} does not contain a SecretString", exc_info=True)
+        logger.error(
+            f"Secret {secret_name} does not contain a SecretString", exc_info=True
+        )
         raise ValueError(f"Secret {secret_name} does not contain a SecretString")
 
     return get_secret_value_response["SecretString"]
@@ -117,6 +121,7 @@ def get_transform_bucket(client=None):
 # convert parquet files into dataframes
 # return a dictionary of dataframes with name as key, and dataframe object as value
 
+
 def get_latest_timestamp(existing_files):
     if existing_files:
         all_datetimes = []
@@ -124,19 +129,17 @@ def get_latest_timestamp(existing_files):
             match = re.search(r"\/(.+/).+_(.+)\.parquet", file_name)
             if match:
                 datetime_str = "".join(match.group(1, 2))
-                all_datetimes.append(
-                    dt.strptime(datetime_str, "%Y/%m/%d/%H:%M:%S")
-                )
+                all_datetimes.append(dt.strptime(datetime_str, "%Y/%m/%d/%H:%M:%S"))
         return max(all_datetimes) if all_datetimes else dt.min
     return existing_files
 
+
 def convert_parquet_files_to_dfs(bucket_name=None, client=None):
     mutable_df_dict = [
         "dim_currency",
         "fact_sales_order",
         "fact_purchase_order",
-        "fact_payment"
-        
+        "fact_payment",
     ]
 
     try:
@@ -145,12 +148,12 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
         if bucket_name is None:
             bucket_name = get_transform_bucket()
         files = client.list_objects_v2(Bucket=bucket_name)
-        
+
         dfs = {}
         if "Contents" in files:
-            s3_key_list = [file["Key"]for file in files["Contents"]]
+            s3_key_list = [file["Key"] for file in files["Contents"]]
             immutables_l = []
-            mutables_d = {prefix:[] for prefix in mutable_df_dict}
+            mutables_d = {prefix: [] for prefix in mutable_df_dict}
             for tab, s3_key in mutables_d.items():
                 for file in s3_key_list:
                     if tab in file:
@@ -161,22 +164,31 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None):
                         continue
             immutables_l = list(set(immutables_l))
             latest_s3_keys = []
-            for k,v in mutables_d.items():
-	            latest_s3_keys.append(dt.strftime(get_latest_timestamp(v), f"{k}/%Y/%m/%d/{k}_%H:%M:%S.parquet"))
-            for file_key in immutables_l+latest_s3_keys:
+            for k, v in mutables_d.items():
+                latest_s3_keys.append(
+                    dt.strftime(
+                        get_latest_timestamp(v), f"{k}/%Y/%m/%d/{k}_%H:%M:%S.parquet"
+                    )
+                )
+            for file_key in immutables_l + latest_s3_keys:
                 try:
                     file_obj = client.get_object(Bucket=bucket_name, Key=file_key)
                     parquet_file = pq.ParquetFile(BytesIO(file_obj["Body"].read()))
                     df = parquet_file.read().to_pandas()
-                    df_without_nulls = df.dropna(how='all') #>> can't do 'any' (default) because we lose rows in dim_location
-                    #print("df_without_nulls", df_without_nulls)
-                    #print("type", type(df_without_nulls))
-                    #print(df_without_nulls.columns)
+                    # >> can't do 'any' (default) because we lose rows in dim_location
+                    df_without_nulls = df.dropna(how="all")
+                    # print("df_without_nulls", df_without_nulls)
+                    # print("type", type(df_without_nulls))
+                    # print(df_without_nulls.columns)
                     dfs[file_key] = df_without_nulls
                 except ClientError as e:
-                    logger.error(f"Unable to retrieve S3 object {file_key}: {e}", exc_info=True)
+                    logger.error(
+                        f"Unable to retrieve S3 object {file_key}: {e}", exc_info=True
+                    )
                 except Exception as e:
-                    logger.error(f"Unable to process file {file_key}: {e}", exc_info=True)
+                    logger.error(
+                        f"Unable to process file {file_key}: {e}", exc_info=True
+                    )
         else:
             logger.error(f"No files found in {bucket_name}.", exc_info=True)
             return {}
@@ -199,23 +211,22 @@ def upload_dfs_to_database():
         "dim_location.parquet",
         "dim_staff.parquet",
         "dim_design.parquet",
-        'dim_transaction.parquet', #This one was missing,
-        'dim_payment_type.parquet'
+        "dim_transaction.parquet",  # This one was missing,
+        "dim_payment_type.parquet",
     ]
     mutable_df_dict = [
         "dim_currency",
         "fact_sales_order",
         "fact_purchase_order",
-        "fact_payment"
-        
+        "fact_payment",
     ]
     with db_engine.begin() as connection:
         for file_name, df in dict_of_dfs.items():
             print(df.dtypes, "dtypes")
             print(df.head())
-            print(file_name,"<<< FILE NAME")
-            print(immutable_df_dict,"<<<IMMUTABLE_DF_DICT")
-            if  file_name in immutable_df_dict: 
+            print(file_name, "<<< FILE NAME")
+            print(immutable_df_dict, "<<<IMMUTABLE_DF_DICT")
+            if file_name in immutable_df_dict:
                 table_name = file_name.split(".")[0]
                 print(table_name, "<<<<<")
                 try:
@@ -229,7 +240,10 @@ def upload_dfs_to_database():
                     upload_status["uploaded"].append(table_name)
                     print(upload_status)
                 except Exception as e:
-                    logger.error(f"Error uploading dataframe {file_name} to database: {e}", exc_info=True)
+                    logger.error(
+                        f"Error uploading dataframe {file_name} to database: {e}",
+                        exc_info=True,
+                    )
                     raise
             elif file_name.split("/")[0] in mutable_df_dict:
                 table_name = file_name.split("/")[0]
@@ -244,11 +258,17 @@ def upload_dfs_to_database():
                     )
                     upload_status["uploaded"].append(table_name)
                 except Exception as e:
-                    logger.error(f"Error uploading dataframe {file_name} to database: {e}", exc_info=True)
+                    logger.error(
+                        f"Error uploading dataframe {file_name} to database: {e}",
+                        exc_info=True,
+                    )
                     raise
             else:
                 upload_status["not_uploaded"].append(file_name)
-                logger.error(f"{file_name} does not correspond with table in database", exc_info=True)
+                logger.error(
+                    f"{file_name} does not correspond with table in database",
+                    exc_info=True,
+                )
             print(upload_status)
     db_engine.dispose()
     return upload_status
diff --git a/src/transform_lambda/dataframes.py b/src/transform_lambda/dataframes.py
index c823b87..6de58e7 100644
--- a/src/transform_lambda/dataframes.py
+++ b/src/transform_lambda/dataframes.py
@@ -36,7 +36,8 @@ def create_fact_sales_order(dict_of_df):
     df_sales["agreed_payment_date"] = pd.to_datetime(
         df_sales["agreed_payment_date"], format="%Y-%m-%d"
     )
-    fact_sales = df_sales.loc[:,
+    fact_sales = df_sales.loc[
+        :,
         [
             "sales_order_id",
             "created_date",
@@ -51,7 +52,7 @@ def create_fact_sales_order(dict_of_df):
             "design_id",
             "agreed_payment_date",
             "agreed_delivery_date",
-            "agreed_delivery_location_id"
+            "agreed_delivery_location_id",
         ],
     ]
     fact_sales.convert_dtypes()
@@ -81,24 +82,24 @@ def create_fact_purchase_orders(dict_of_df):
     df_po["agreed_payment_date"] = pd.to_datetime(
         df_po["agreed_payment_date"], format="%Y-%m-%d"
     )
-    fact_purchase_order = df_po.loc[:,
-        [                           
-        "purchase_order_id",
-        "created_date",
-        "created_time",
-        "last_updated_date",
-        "last_updated_time",
-        "staff_id",
-        "counterparty_id",
-        "item_code",
-        "item_quantity",
-        "item_unit_price",
-        "currency_id",
-        "agreed_delivery_date",
-        "agreed_payment_date",
-        "agreed_delivery_location_id"
-        ]
-
+    fact_purchase_order = df_po.loc[
+        :,
+        [
+            "purchase_order_id",
+            "created_date",
+            "created_time",
+            "last_updated_date",
+            "last_updated_time",
+            "staff_id",
+            "counterparty_id",
+            "item_code",
+            "item_quantity",
+            "item_unit_price",
+            "currency_id",
+            "agreed_delivery_date",
+            "agreed_payment_date",
+            "agreed_delivery_location_id",
+        ],
     ]
     fact_purchase_order.convert_dtypes()
     fact_purchase_order.index = pd.RangeIndex(1, len(fact_purchase_order.index) + 1)
@@ -128,28 +129,29 @@ def create_fact_payment(dict_of_df):
     df_payment["payment_date"] = pd.to_datetime(
         df_payment["payment_date"], format="%Y-%m-%d"
     )
-    fact_payment = df_payment.loc[:,
+    fact_payment = df_payment.loc[
+        :,
         [
-        "payment_id",
-        "created_date",
-        "created_time",
-        "last_updated_date",
-        "last_updated_time",
-        "transaction_id",
-        "counterparty_id",
-        "payment_amount",
-        "currency_id",
-        "payment_type_id",
-        "paid",
-        "payment_date"
-        ]
+            "payment_id",
+            "created_date",
+            "created_time",
+            "last_updated_date",
+            "last_updated_time",
+            "transaction_id",
+            "counterparty_id",
+            "payment_amount",
+            "currency_id",
+            "payment_type_id",
+            "paid",
+            "payment_date",
+        ],
     ]
     fact_payment.convert_dtypes()
     fact_payment.index = pd.RangeIndex(1, len(fact_payment.index) + 1)
     fact_payment.index.name = "payment_record_id"
     fact_payment.reset_index(inplace=True)
     fact_payment.dropna(inplace=True)
-    fact_payment = fact_payment.astype({'currency_id':'int','payment_id':'int'})
+    fact_payment = fact_payment.astype({"currency_id": "int", "payment_id": "int"})
     return fact_payment
 
 
@@ -157,15 +159,10 @@ def create_fact_payment(dict_of_df):
 
 
 def create_dim_transaction(dict_of_df):
-    dim_transaction = dict_of_df["transaction"].loc[:,
-        [
-        "transaction_id",
-        "transaction_type",
-        "sales_order_id",
-        "purchase_order_id"
-        ]
+    dim_transaction = dict_of_df["transaction"].loc[
+        :, ["transaction_id", "transaction_type", "sales_order_id", "purchase_order_id"]
     ]
-    #dim_transaction = dim_transaction.astype({"sales_order_id":"Int64","purchase_order_id":"Int64"})
+    # dim_transaction = dim_transaction.astype({"sales_order_id":"Int64","purchase_order_id":"Int64"})
     return dim_transaction
 
 
@@ -174,7 +171,8 @@ def create_dim_transaction(dict_of_df):
 
 def create_dim_location(dict_of_df):
     dim_location = (
-        dict_of_df["address"].drop(labels=["created_at", "last_updated"], axis=1)
+        dict_of_df["address"]
+        .drop(labels=["created_at", "last_updated"], axis=1)
         .rename(columns={"address_id": "location_id"})
     )
     return dim_location
@@ -193,7 +191,7 @@ def create_dim_counterparty(dict_of_df):
         left_on="legal_address_id",
         right_on="counterparty_legal_address_id",
         how="inner",
-    )#.dropna(inplace=True)
+    )  # .dropna(inplace=True)
     dim_counterparty = df_cp.drop(
         labels=[
             "legal_address_id",
@@ -201,8 +199,9 @@ def create_dim_counterparty(dict_of_df):
             "created_at",
             "last_updated",
             "commercial_contact",
-            "delivery_contact"
-        ], axis=1
+            "delivery_contact",
+        ],
+        axis=1,
     )
     return dim_counterparty
 
@@ -272,12 +271,7 @@ def create_dim_currency(dict_of_df, names=scrape_currency_names()):
 
 def create_dim_payment_type(dict_of_df):
     df_payment_type = dict_of_df["payment_type"]
-    dim_payment_type = df_payment_type.loc[:, 
-        [
-            "payment_type_id", 
-            "payment_type_name"
-        ]
-    ]
+    dim_payment_type = df_payment_type.loc[:, ["payment_type_id", "payment_type_name"]]
     return dim_payment_type
 
 
@@ -286,13 +280,8 @@ def create_dim_payment_type(dict_of_df):
 
 def create_dim_design(dict_of_df):
     df_design = dict_of_df["design"]
-    dim_design = df_design.loc[:, 
-        [
-            "design_id", 
-            "design_name", 
-            "file_name", 
-            "file_location"
-        ]
+    dim_design = df_design.loc[
+        :, ["design_id", "design_name", "file_name", "file_location"]
     ]
     return dim_design
 
@@ -304,14 +293,15 @@ def create_dim_staff(dict_of_df):
     staff_department = pd.merge(
         dict_of_df["staff"], dict_of_df["department"], on="department_id", how="left"
     )
-    dim_staff = staff_department.loc[:,
+    dim_staff = staff_department.loc[
+        :,
         [
             "staff_id",
             "first_name",
             "last_name",
             "department_name",
             "location",
-            "email_address"
-        ]
+            "email_address",
+        ],
     ]
     return dim_staff
diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py
index 5ea8cf0..2739997 100644
--- a/src/transform_lambda/transform_lambda.py
+++ b/src/transform_lambda/transform_lambda.py
@@ -42,7 +42,7 @@ TABLES = [
     "department",
     "currency",
     "design",
-    "payment_type"
+    "payment_type",
 ]
 
 
@@ -66,14 +66,14 @@ def lambda_handler(event, context):
             "dim_staff": create_dim_staff(dict_of_df),
             "dim_design": create_dim_design(dict_of_df),
             "dim_transaction": create_dim_transaction(dict_of_df),
-            "dim_payment_type": create_dim_payment_type(dict_of_df)
+            "dim_payment_type": create_dim_payment_type(dict_of_df),
         }
 
         mutable_df_dict = {
             "fact_sales_order": create_fact_sales_order(dict_of_df),
             "fact_purchase_order": create_fact_purchase_orders(dict_of_df),
             "fact_payment": create_fact_payment(dict_of_df),
-            "dim_currency": create_dim_currency(dict_of_df)
+            "dim_currency": create_dim_currency(dict_of_df),
         }
         print(immutable_df_dict.values())
         print(mutable_df_dict.values())
-- 
cgit v1.2.3