From 3f40e96217418590ca66af6912f595cc04425849 Mon Sep 17 00:00:00 2001 From: lian-manonog Date: Mon, 19 Aug 2024 15:52:14 +0100 Subject: wip: setting up test files for transform_lambda --- tests/test_transform_lambda.py | 1 + 1 file changed, 1 insertion(+) create mode 100644 tests/test_transform_lambda.py (limited to 'tests/test_transform_lambda.py') diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py new file mode 100644 index 0000000..dd08b6a --- /dev/null +++ b/tests/test_transform_lambda.py @@ -0,0 +1 @@ +from src.transform_lambda import lambda_handler \ No newline at end of file -- cgit v1.2.3 From b4fafcd9731f11f6f2efde843242b9c5cb84e85f Mon Sep 17 00:00:00 2001 From: Ang Bel Date: Wed, 21 Aug 2024 12:50:32 +0100 Subject: function to write files from s3 into a list of dataframes. Current test is failing due to AioClientCreator object has no attribute "_inject_s3_input_parameters" --- requirements.txt | 2 +- src/transform_lambda.py | 34 ++++++++++++++++++++++++++++++---- tests/test_transform_lambda.py | 34 +++++++++++++++++++++++++++++++++- 3 files changed, 64 insertions(+), 6 deletions(-) (limited to 'tests/test_transform_lambda.py') diff --git a/requirements.txt b/requirements.txt index 6f383f9..087d1c2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,6 @@ asn1crypto==1.5.1 boto3==1.34.159 -botocore==1.34.159 +botocore==1.34.7 certifi==2024.7.4 cffi==1.17.0 charset-normalizer==3.3.2 diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 900bf4b..6f65728 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -1,3 +1,4 @@ +#from src.extract_lambda import extract_bucket import json import boto3 import re @@ -10,9 +11,7 @@ import pandas as pd ##In-order to use PANDAS module in lambda function, a Lambda Layer needs to be attached to the AWS Lambda Function. ##need a function that normalises the data - - -s3_resource = boto3.resource('s3') ##need this for a way of reuploading data after transformation +#s3_resource = boto3.resource('s3') ##need this for a way of reuploading data after transformation def lambda_handler(event, context): s3_client = boto3.client('s3') @@ -54,4 +53,31 @@ def lambda_handler(event, context): ## each csv file must be converted into a pandas df ## done via read_csv, where stringIO creates an file-like-object from string - treats string like a file: as file is not physically stored in file -## each file needs its own panda df (?) to be normalised \ No newline at end of file +## each file needs its own panda df (?) to be normalised +tables = ['sales_order', + 'transaction', + 'payment', + 'counterparty', + 'address', + 'staff', + 'purchase_order', + 'department', + 'currency', + 'design', + 'payment_type'] + +def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client('s3')): + table_dfs = {} + for table in tables: + response = client.list_objects_v2(Bucket=bucket, Prefix=table) + list_of_keys = ['s3://'+object['Key'] for object in response['Contents']] + print(list_of_keys) + list_of_df = [pd.read_csv(key) for key in list_of_keys] + table_dfs[table] = pd.concat(list_of_df) + return table_dfs + # exec("%s = %d" % (table,pd.concat(list_of_df))) + # exec(f"{table} = {pd.concat(list_of_df)}") + # table_dfs = [sales_order, transaction, payment, counterparty, address, + # staff, purchase_order, department, currency, design, payment_type] + + diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index dd08b6a..a3ec4a8 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -1 +1,33 @@ -from src.transform_lambda import lambda_handler \ No newline at end of file +from src.transform_lambda import read_from_s3_subfolder_to_df +from moto import mock_aws +import pytest +import pandas as pd +import os +import boto3 + +@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') +class TestReadFromS3: + + def test_returns_dictionary_with_correct_value_pair(self,s3_client): + s3_client.create_bucket(Bucket = 'dummy_buc',CreateBucketConfiguration={ + 'LocationConstraint': 'eu-west-2' + }) + s3_client.upload_file('tests/dummy_identical.csv', 'dummy_buc', 'Foods/2024/08/21/Foods_12:03:10.csv') + tables = ['Foods'] + result = read_from_s3_subfolder_to_df(tables,bucket='dummy_buc',client=s3_client) + print(result) + assert isinstance(result,dict) + assert list(result.keys()) == 'Foods' + assert isinstance(result['Foods'],pd.DataFrame) + \ No newline at end of file -- cgit v1.2.3 From 0c6e2f8486d1ec4d9b0bd4984e01baca3a159df0 Mon Sep 17 00:00:00 2001 From: Ang Bel Date: Wed, 21 Aug 2024 15:07:51 +0100 Subject: (tests) Read from s3 to df passes --- src/transform_lambda.py | 26 ++++---------------------- tests/dummy_2.csv | 5 +++++ tests/test_transform_lambda.py | 21 +++++++++++++++++++-- 3 files changed, 28 insertions(+), 24 deletions(-) create mode 100644 tests/dummy_2.csv (limited to 'tests/test_transform_lambda.py') diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 6f65728..ea4e16f 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -15,18 +15,6 @@ import pandas as pd def lambda_handler(event, context): s3_client = boto3.client('s3') - - tables = ['sales_order', - 'transaction', - 'payment', - 'counterparty', - 'address', - 'staff', - 'purchase_order', - 'department', - 'currency', - 'design', - 'payment_type'] try: s3_bucket_name = event["Records"][0]["s3"]["bucket"]["name"] s3_file_name = event["Records"][0]["s3"]["object"]["key"] @@ -51,9 +39,8 @@ def lambda_handler(event, context): 'body': json.dumps('') } -## each csv file must be converted into a pandas df -## done via read_csv, where stringIO creates an file-like-object from string - treats string like a file: as file is not physically stored in file -## each file needs its own panda df (?) to be normalised +## Started from fresh on Wed 21st Aug: + tables = ['sales_order', 'transaction', 'payment', @@ -70,14 +57,9 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client('s3')): table_dfs = {} for table in tables: response = client.list_objects_v2(Bucket=bucket, Prefix=table) - list_of_keys = ['s3://'+object['Key'] for object in response['Contents']] - print(list_of_keys) + list_of_keys = ['s3://'+bucket+'/'+object['Key'] for object in response['Contents']] list_of_df = [pd.read_csv(key) for key in list_of_keys] table_dfs[table] = pd.concat(list_of_df) return table_dfs - # exec("%s = %d" % (table,pd.concat(list_of_df))) - # exec(f"{table} = {pd.concat(list_of_df)}") - # table_dfs = [sales_order, transaction, payment, counterparty, address, - # staff, purchase_order, department, currency, design, payment_type] - + diff --git a/tests/dummy_2.csv b/tests/dummy_2.csv new file mode 100644 index 0000000..8abc9bf --- /dev/null +++ b/tests/dummy_2.csv @@ -0,0 +1,5 @@ +Car_type,Brand,Colour +Truck,Chevrolet,Grey +Convertible,Mercedes,Red +Van,Volkswagen,Blue + diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index a3ec4a8..7de1bf3 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -4,6 +4,7 @@ import pytest import pandas as pd import os import boto3 +import numpy as np @pytest.fixture(scope='class') def aws_credentials(): @@ -27,7 +28,23 @@ class TestReadFromS3: tables = ['Foods'] result = read_from_s3_subfolder_to_df(tables,bucket='dummy_buc',client=s3_client) print(result) + expected_df = pd.DataFrame(np.array([['Vegetable', 'Sour', 'Green'], ['Berry', 'Sweet', 'Red']]), + columns=['Food_type', 'Flavour', 'Colour']) assert isinstance(result,dict) - assert list(result.keys()) == 'Foods' + assert list(result.keys())[0] == 'Foods' assert isinstance(result['Foods'],pd.DataFrame) - \ No newline at end of file + assert result['Foods'].eq(expected_df,axis='columns').all(axis=None) + + def test_returns_dictionary_of_dataframes_for_multiple_tables(self,s3_client): + s3_client.upload_file('tests/dummy_2.csv', 'dummy_buc', 'Cars/2024/08/21/Cars_14:03:56.csv') + tables = ['Foods','Cars'] + result = read_from_s3_subfolder_to_df(tables,bucket='dummy_buc',client=s3_client) + expected_foods_df = pd.DataFrame(np.array([['Vegetable', 'Sour', 'Green'], ['Berry', 'Sweet', 'Red']]), + columns=['Food_type', 'Flavour', 'Colour']) + expected_cars_df = pd.DataFrame(np.array([['Truck', 'Chevrolet', 'Grey'], ['Convertible', 'Mercedes','Red'],['Van','Volkswagen','Blue']]), + columns=['Car_type', 'Brand', 'Colour']) + assert list(result.keys()) == tables + assert result['Foods'].eq(expected_foods_df,axis='columns').all(axis=None) + assert result['Cars'].eq(expected_cars_df,axis='columns').all(axis=None) + + -- cgit v1.2.3 From c8e94530b65d6807b2b9bb246a542963839cce9d Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Wed, 21 Aug 2024 14:49:56 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in b882bb0 according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/84 --- src/transform_lambda.py | 36 +++++++++------- tests/test_transform_lambda.py | 94 ++++++++++++++++++++++++++---------------- 2 files changed, 79 insertions(+), 51 deletions(-) (limited to 'tests/test_transform_lambda.py') diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 3a7cf43..b176ccc 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -1,4 +1,4 @@ -#from src.extract_lambda import extract_bucket +# from src.extract_lambda import extract_bucket import json import boto3 import re @@ -6,29 +6,33 @@ import io from io import StringIO import pandas as pd + def lambda_handler(event, context): pass -tables = ['sales_order', - 'transaction', - 'payment', - 'counterparty', - 'address', - 'staff', - 'purchase_order', - 'department', - 'currency', - 'design', - 'payment_type'] +tables = [ + "sales_order", + "transaction", + "payment", + "counterparty", + "address", + "staff", + "purchase_order", + "department", + "currency", + "design", + "payment_type", +] + -def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client('s3')): +def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")): table_dfs = {} for table in tables: response = client.list_objects_v2(Bucket=bucket, Prefix=table) - list_of_keys = ['s3://'+bucket+'/'+object['Key'] for object in response['Contents']] + list_of_keys = [ + "s3://" + bucket + "/" + object["Key"] for object in response["Contents"] + ] list_of_df = [pd.read_csv(key) for key in list_of_keys] table_dfs[table] = pd.concat(list_of_df) return table_dfs - - diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 7de1bf3..5121905 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -6,45 +6,69 @@ import os import boto3 import numpy as np -@pytest.fixture(scope='class') + +@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' + 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') +@pytest.fixture(scope="class") def s3_client(aws_credentials): with mock_aws(): - yield boto3.client('s3') + yield boto3.client("s3") + + class TestReadFromS3: - - def test_returns_dictionary_with_correct_value_pair(self,s3_client): - s3_client.create_bucket(Bucket = 'dummy_buc',CreateBucketConfiguration={ - 'LocationConstraint': 'eu-west-2' - }) - s3_client.upload_file('tests/dummy_identical.csv', 'dummy_buc', 'Foods/2024/08/21/Foods_12:03:10.csv') - tables = ['Foods'] - result = read_from_s3_subfolder_to_df(tables,bucket='dummy_buc',client=s3_client) + def test_returns_dictionary_with_correct_value_pair(self, s3_client): + s3_client.create_bucket( + Bucket="dummy_buc", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + s3_client.upload_file( + "tests/dummy_identical.csv", + "dummy_buc", + "Foods/2024/08/21/Foods_12:03:10.csv", + ) + tables = ["Foods"] + result = read_from_s3_subfolder_to_df( + tables, bucket="dummy_buc", client=s3_client + ) print(result) - expected_df = pd.DataFrame(np.array([['Vegetable', 'Sour', 'Green'], ['Berry', 'Sweet', 'Red']]), - columns=['Food_type', 'Flavour', 'Colour']) - assert isinstance(result,dict) - assert list(result.keys())[0] == 'Foods' - assert isinstance(result['Foods'],pd.DataFrame) - assert result['Foods'].eq(expected_df,axis='columns').all(axis=None) - - def test_returns_dictionary_of_dataframes_for_multiple_tables(self,s3_client): - s3_client.upload_file('tests/dummy_2.csv', 'dummy_buc', 'Cars/2024/08/21/Cars_14:03:56.csv') - tables = ['Foods','Cars'] - result = read_from_s3_subfolder_to_df(tables,bucket='dummy_buc',client=s3_client) - expected_foods_df = pd.DataFrame(np.array([['Vegetable', 'Sour', 'Green'], ['Berry', 'Sweet', 'Red']]), - columns=['Food_type', 'Flavour', 'Colour']) - expected_cars_df = pd.DataFrame(np.array([['Truck', 'Chevrolet', 'Grey'], ['Convertible', 'Mercedes','Red'],['Van','Volkswagen','Blue']]), - columns=['Car_type', 'Brand', 'Colour']) - assert list(result.keys()) == tables - assert result['Foods'].eq(expected_foods_df,axis='columns').all(axis=None) - assert result['Cars'].eq(expected_cars_df,axis='columns').all(axis=None) - + expected_df = pd.DataFrame( + np.array([["Vegetable", "Sour", "Green"], ["Berry", "Sweet", "Red"]]), + columns=["Food_type", "Flavour", "Colour"], + ) + assert isinstance(result, dict) + assert list(result.keys())[0] == "Foods" + assert isinstance(result["Foods"], pd.DataFrame) + assert result["Foods"].eq(expected_df, axis="columns").all(axis=None) + def test_returns_dictionary_of_dataframes_for_multiple_tables(self, s3_client): + s3_client.upload_file( + "tests/dummy_2.csv", "dummy_buc", "Cars/2024/08/21/Cars_14:03:56.csv" + ) + tables = ["Foods", "Cars"] + result = read_from_s3_subfolder_to_df( + tables, bucket="dummy_buc", client=s3_client + ) + expected_foods_df = pd.DataFrame( + np.array([["Vegetable", "Sour", "Green"], ["Berry", "Sweet", "Red"]]), + columns=["Food_type", "Flavour", "Colour"], + ) + expected_cars_df = pd.DataFrame( + np.array( + [ + ["Truck", "Chevrolet", "Grey"], + ["Convertible", "Mercedes", "Red"], + ["Van", "Volkswagen", "Blue"], + ] + ), + columns=["Car_type", "Brand", "Colour"], + ) + assert list(result.keys()) == tables + assert result["Foods"].eq(expected_foods_df, axis="columns").all(axis=None) + assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None) -- cgit v1.2.3 From 2238618164eb838c8b5e27c2cf3f5ed748637a3d Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 22 Aug 2024 12:17:18 +0100 Subject: chore: skip transform_lambda tests are they are broken --- tests/test_transform_lambda.py | 2 ++ 1 file changed, 2 insertions(+) (limited to 'tests/test_transform_lambda.py') diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 5121905..4c689f7 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -23,6 +23,7 @@ def s3_client(aws_credentials): class TestReadFromS3: + @pytest.mark.skip(reason="The test is broken!") def test_returns_dictionary_with_correct_value_pair(self, s3_client): s3_client.create_bucket( Bucket="dummy_buc", @@ -47,6 +48,7 @@ class TestReadFromS3: assert isinstance(result["Foods"], pd.DataFrame) assert result["Foods"].eq(expected_df, axis="columns").all(axis=None) + @pytest.mark.skip(reason="The test is broken!") def test_returns_dictionary_of_dataframes_for_multiple_tables(self, s3_client): s3_client.upload_file( "tests/dummy_2.csv", "dummy_buc", "Cars/2024/08/21/Cars_14:03:56.csv" -- cgit v1.2.3 From daee22145e8ce27425dd8de941b5ab65e6a619ae Mon Sep 17 00:00:00 2001 From: lian-manonog Date: Thu, 22 Aug 2024 16:03:16 +0100 Subject: Refactored tests for transform lambda - all passing now --- tests/test_transform_lambda.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) (limited to 'tests/test_transform_lambda.py') diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 5121905..516f83b 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -39,8 +39,8 @@ class TestReadFromS3: ) print(result) expected_df = pd.DataFrame( - np.array([["Vegetable", "Sour", "Green"], ["Berry", "Sweet", "Red"]]), - columns=["Food_type", "Flavour", "Colour"], + np.array([["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"]]), + columns=["Food_type", "Flavour", "Colour", "last_updated"], ) assert isinstance(result, dict) assert list(result.keys())[0] == "Foods" @@ -56,8 +56,8 @@ class TestReadFromS3: tables, bucket="dummy_buc", client=s3_client ) expected_foods_df = pd.DataFrame( - np.array([["Vegetable", "Sour", "Green"], ["Berry", "Sweet", "Red"]]), - columns=["Food_type", "Flavour", "Colour"], + np.array([["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"]]), + columns=["Food_type", "Flavour", "Colour", "last_updated"], ) expected_cars_df = pd.DataFrame( np.array( @@ -72,3 +72,5 @@ class TestReadFromS3: assert list(result.keys()) == tables assert result["Foods"].eq(expected_foods_df, axis="columns").all(axis=None) assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None) + + -- cgit v1.2.3 From 2231ea89329bd500f7371b7395f5208f7a86c20e Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Fri, 23 Aug 2024 10:11:40 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 8e20c5c according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/93 --- src/dataframes.py | 293 +++++++++++++++++++++++++---------------- src/transform_lambda.py | 100 +++++++------- tests/test_fact_sales_order.py | 90 ++++++++++--- tests/test_transform_lambda.py | 16 ++- 4 files changed, 319 insertions(+), 180 deletions(-) (limited to 'tests/test_transform_lambda.py') diff --git a/src/dataframes.py b/src/dataframes.py index 9ce3be0..684f102 100644 --- a/src/dataframes.py +++ b/src/dataframes.py @@ -8,7 +8,7 @@ import re from datetime import datetime as dt import requests -#Table names: +# Table names: # fact_sales_order # fact_purchase_orders # fact_payment @@ -21,9 +21,11 @@ import requests # dim_currency # dim_counterparty + def create_dim_transaction(dict_of_df): pass + def create_fact_sales_order(dict_of_df): df_sales = dict_of_df["sales_order"] df_sales.index.name = "sales_record_id" @@ -33,36 +35,46 @@ def create_fact_sales_order(dict_of_df): df_sales["last_updated_time"] = pd.to_datetime(df_sales["last_updated"]).dt.time pd.merge(dict_of_df["staff"], df_sales["sales_staff_id"], on="staff_id", how="left") # df_sales.rename(columns={"staff_id": "sales_staff_id"}) - fact_sales_order = df_sales.loc[:,[ - "sales_record_id", - "sales_order_id", - "created_date", - "created_time", - "last_updated_date", - "last_updated_time", - "sales_staff_id", - "counterparty_id", - "units_sold", - "unit_price", - "currency_id", - "design_id", - "agreed_payment_date", - "agreed_delivery_date", - "agreed_delivery_location_id" - ]] + fact_sales_order = df_sales.loc[ + :, + [ + "sales_record_id", + "sales_order_id", + "created_date", + "created_time", + "last_updated_date", + "last_updated_time", + "sales_staff_id", + "counterparty_id", + "units_sold", + "unit_price", + "currency_id", + "design_id", + "agreed_payment_date", + "agreed_delivery_date", + "agreed_delivery_location_id", + ], + ] return fact_sales_order -## fact_purchase_order from purchase_order + +# fact_purchase_order from purchase_order + + 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'].date() - df_po['created_time'] = df_po['created_at'].dt.time - df_po['last_updated_date'] = df_po['last_updated_at'].date() - df_po['last_updated_time'] = df_po['last_updated_at'].dt.time - df_po['agreed_delivery_date'] = pd.to_datetime(df_po['agreed_delivery_date'],format="%Y-%m-%d") - df_po['agreed_payment_date'] = pd.to_datetime(df_po['agreed_payment_date'],format="%Y-%m-%d") - df_po.drop(labels=['created_at','last_updated_at'],axis=1,inplace=True) + df_po = dict_of_df["purchase_order"] + df_po.index.name = "purchase_record_id" + df_po["created_date"] = df_po["created_at"].date() + df_po["created_time"] = df_po["created_at"].dt.time + df_po["last_updated_date"] = df_po["last_updated_at"].date() + df_po["last_updated_time"] = df_po["last_updated_at"].dt.time + df_po["agreed_delivery_date"] = pd.to_datetime( + df_po["agreed_delivery_date"], format="%Y-%m-%d" + ) + df_po["agreed_payment_date"] = pd.to_datetime( + df_po["agreed_payment_date"], format="%Y-%m-%d" + ) + df_po.drop(labels=["created_at", "last_updated_at"], axis=1, inplace=True) return df_po @@ -73,69 +85,97 @@ def create_fact_payment(dict_of_df): df_payment["created_time"] = pd.to_datetime(df_payment["created_at"]).dt.time df_payment["last_updated_date"] = pd.to_datetime(df_payment["last_updated"]).dt.date df_payment["last_updated_time"] = pd.to_datetime(df_payment["last_updated"]).dt.time - fact_payment = df_payment.loc[:,[ - "payment_record_id", - "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 = df_payment.loc[ + :, + [ + "payment_record_id", + "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", + ], + ] return fact_payment -## dim_location from address --> drops 2 columns + +# dim_location from address --> drops 2 columns + + def create_dim_location(dict_of_df): - df_loc = dict_of_df['address'].drop(labels=['created_at', 'last_updated'], axis=1).rename(columns={'address_id': 'location_id'}).set_index('location_id') + df_loc = ( + dict_of_df["address"] + .drop(labels=["created_at", "last_updated"], axis=1) + .rename(columns={"address_id": "location_id"}) + .set_index("location_id") + ) return df_loc -## dim_counterparty from address and counterparty + +# dim_counterparty from address and counterparty + + def create_dim_counterparty(dict_of_df): - df_prefixed_address = dict_of_df['address'].add_prefix('counterparty_legal_', axis=1) - df_cp = pd.merge(dict_of_df['counterparty'], - df_prefixed_address, - left_on="legal_address_id", - right_on="address_id", - how="outer").set_index('counterparty_id') + df_prefixed_address = dict_of_df["address"].add_prefix( + "counterparty_legal_", axis=1 + ) + df_cp = pd.merge( + dict_of_df["counterparty"], + df_prefixed_address, + left_on="legal_address_id", + right_on="address_id", + how="outer", + ).set_index("counterparty_id") return df_cp -## dim_date from purchase_order +# dim_date from purchase_order def create_dim_date(dict_of_df): - sr_date = pd.concat([dict_of_df['created_date'],dict_of_df['last_updated_date'],dict_of_df['agreed_delivery_date'],dict_of_df['agreed_payment_date']]).sort() - df_date = pd.DataFrame(sr_date,columns='date_id') - 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 - df_date['day_of_week'] = df_date['date_id'].dt.dayofweek - df_date['day_name'] = df_date['date_id'].dt.day_name - df_date['month_name'] = df_date['date_id'].dt.month_name - df_date['quarter'] = df_date['date_id'].dt.quarter - df_date.set_index('date_id') + sr_date = pd.concat( + [ + dict_of_df["created_date"], + dict_of_df["last_updated_date"], + dict_of_df["agreed_delivery_date"], + dict_of_df["agreed_payment_date"], + ] + ).sort() + df_date = pd.DataFrame(sr_date, columns="date_id") + 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 + df_date["day_of_week"] = df_date["date_id"].dt.dayofweek + df_date["day_name"] = df_date["date_id"].dt.day_name + df_date["month_name"] = df_date["date_id"].dt.month_name + df_date["quarter"] = df_date["date_id"].dt.quarter + df_date.set_index("date_id") + def scrape_currency_names(): - response = requests.get('https://www.xe.com/currency/').content - soup = BeautifulSoup(response,'html.parser') - currency = [item.text for item in soup.findAll('a', attrs={'class' : "sc-299dec64-6 fZPTSw"})] + response = requests.get("https://www.xe.com/currency/").content + soup = BeautifulSoup(response, "html.parser") + currency = [ + item.text for item in soup.findAll("a", attrs={"class": "sc-299dec64-6 fZPTSw"}) + ] sr = pd.Series(currency) - df_cur = sr.str.split(pat=" - ",expand=True).rename({0:'currency_code',1:'currency_name'},axis=1) + df_cur = sr.str.split(pat=" - ", expand=True).rename( + {0: "currency_code", 1: "currency_name"}, axis=1 + ) return df_cur -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').set_index('currency_id') - return dim_cur - - - - +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" + ).set_index("currency_id") + return dim_cur def create_dim_payment_type(dict_of_df): @@ -143,6 +183,7 @@ def create_dim_payment_type(dict_of_df): dim_payment_type = df_payment_type.loc[:, ["payment_type_id", "payment_type_name"]] return dim_payment_type + def create_fact_payment(dict_of_df): df_payment = dict_of_df["payment"] df_payment.index.name = "payment_record_id" @@ -150,41 +191,57 @@ def create_fact_payment(dict_of_df): df_payment["created_time"] = pd.to_datetime(df_payment["created_at"]).dt.time df_payment["last_updated_date"] = pd.to_datetime(df_payment["last_updated"]).dt.date df_payment["last_updated_time"] = pd.to_datetime(df_payment["last_updated"]).dt.time - fact_payment = df_payment.loc[:,[ - "payment_record_id", - "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 = df_payment.loc[ + :, + [ + "payment_record_id", + "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", + ], + ] return fact_payment + 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 + 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[:, ['staff_id', 'first_name', 'last_name', 'department_name', 'location', 'email_address']] + staff_department = pd.merge( + dict_of_df["staff"], dict_of_df["department"], on="department_id", how="left" + ) + dim_staff = staff_department.loc[ + :, + [ + "staff_id", + "first_name", + "last_name", + "department_name", + "location", + "email_address", + ], + ] return dim_staff + def create_dim_currency(dict_of_df): df_currency = dict_of_df["currency"] dim_currency = df_currency.loc[:, ["currency_id", "currency_code"]] - mappings = { - "GBP": "Pound", - "USD": "US Dollar", - "EUR": "Euro" - } + mappings = {"GBP": "Pound", "USD": "US Dollar", "EUR": "Euro"} dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) return dim_currency @@ -200,39 +257,49 @@ def create_dim_date(dict_of_df): df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() - dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() + dim_date = [ + "date_id", + "year", + "month", + "day", + "day_of_week", + "day_name", + "month_name", + "quarter", + ] # series.dt.quarter() return dim_date -# TO DO: +# TO DO: # complete dim_date from merged fact table # merge dataframes into one dataframe # remove duplicates # test dim_date and fact_sales_order + def create_sales_star_schema(dict_of_df): dim_design = create_dim_design(dict_of_df) dim_staff = create_dim_staff(dict_of_df) dim_currency = create_dim_currency(dict_of_df) dim_date = create_dim_date(dict_of_df) - + fact_sales_order = create_fact_sales_order(dict_of_df) - - fact_sales_order = fact_sales_order.merge(dim_design, on='design_id', how='left') - fact_sales_order = fact_sales_order.merge(dim_staff, left_on='sales_staff_id', right_on='staff_id', how='left') - fact_sales_order = fact_sales_order.merge(dim_currency, on='currency_id', how='left') - fact_sales_order = fact_sales_order.merge(dim_date, left_on='agreed_delivery_date', right_on='date_id', how='left') - - return fact_sales_order + fact_sales_order = fact_sales_order.merge(dim_design, on="design_id", how="left") + fact_sales_order = fact_sales_order.merge( + dim_staff, left_on="sales_staff_id", right_on="staff_id", how="left" + ) + fact_sales_order = fact_sales_order.merge( + dim_currency, on="currency_id", how="left" + ) + fact_sales_order = fact_sales_order.merge( + dim_date, left_on="agreed_delivery_date", right_on="date_id", how="left" + ) + + return fact_sales_order 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"]] return dim_payment_type - - - - - diff --git a/src/transform_lambda.py b/src/transform_lambda.py index d30d91d..3e74ee0 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -6,12 +6,14 @@ import pandas as pd import pyarrow as pa import pyarrow.parquet as pq from src.dataframes import * + # from src.extract_lambda import extract_bucket, DBConnectionException import boto3 from botocore.exceptions import ClientError from pg8000.native import Connection, InterfaceError from datetime import datetime + class DBConnectionException(Exception): """Wraps pg8000.native Error or DatabaseError.""" @@ -20,6 +22,7 @@ class DBConnectionException(Exception): self.message = str(e) super().__init__(self.message) + logger = logging.getLogger(__name__) logging.basicConfig( @@ -45,44 +48,45 @@ tables = [ "payment_type", ] + def lambda_handler(event, context): db = None - - try: + + try: db = connect_to_database() - bucket = bucket_name('transform') + bucket = bucket_name("transform") existing_s3_files = list_existing_s3_files(bucket) - dict_of_df = read_from_s3_subfolder_to_df(tables, extract_bucket(), client=boto3.client("s3")) + dict_of_df = read_from_s3_subfolder_to_df( + tables, extract_bucket(), client=boto3.client("s3") + ) immutable_df_dict = { - 'dim_counterparty': create_dim_counterparty(dict_of_df), - 'dim_date': create_dim_date(dict_of_df), - '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_counterparty": create_dim_counterparty(dict_of_df), + "dim_date": create_dim_date(dict_of_df), + "dim_location": create_dim_location(dict_of_df), + "dim_staff": create_dim_staff(dict_of_df), + "dim_design": create_dim_design(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)} - + "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), + } + status = process_to_parquet_and_upload_to_s3( - existing_s3_files, - immutable_df_dict, - mutable_df_dict, - bucket + existing_s3_files, immutable_df_dict, mutable_df_dict, bucket ) - - if not status['uploaded']: + + if not status["uploaded"]: logger.info("No dataframes written to the bucket.") return { - 'statusCode': 204, - "body": json.dumps("No files where uploaded."), + "statusCode": 204, + "body": json.dumps("No files where uploaded."), } - + return { "statusCode": 200, "body": json.dumps( @@ -90,7 +94,7 @@ def lambda_handler(event, context): 'The following tables were not uploaded: '+', '.join([status['not_uploaded']]) if status['not_uploaded'] else ''}""" ), } - + except Exception as e: logger.error(f"Error: {e}", exc_info=True) return {"statusCode": 500, "body": json.dumps("Internal server error.")} @@ -99,34 +103,38 @@ def lambda_handler(event, context): db.close() -def process_to_parquet_and_upload_to_s3(existing_s3_files, - immutable_df_dict, - mutable_df_dict, - bucket, - client=boto3.client('s3')): - status = {'uploaded': [], - 'not_uploaded': []} +def process_to_parquet_and_upload_to_s3( + existing_s3_files, + immutable_df_dict, + mutable_df_dict, + bucket, + client=boto3.client("s3"), +): + status = {"uploaded": [], "not_uploaded": []} for table_name, df in immutable_df_dict.items(): if table_name in existing_s3_files: - status['not_uploaded'].append(table_name) + status["not_uploaded"].append(table_name) else: - parquet_file = df.to_parquet(f'{table_name}.parquet', engine='pyarrow') #or fastparquet - client.upload_file(parquet_file, bucket, f'{table_name}.parquet') - status['uploaded'].append(table_name) + parquet_file = df.to_parquet( + f"{table_name}.parquet", engine="pyarrow" + ) # or fastparquet + client.upload_file(parquet_file, bucket, f"{table_name}.parquet") + status["uploaded"].append(table_name) for table_name, df in mutable_df_dict.items(): s3_key = datetime.strftime( - datetime.today(), f"{table_name}/%Y/%m/%d/{table_name}_%H:%M:%S.parquet") - parquet_file = df.to_parquet(f'{table_name}.parquet', engine='pyarrow') #or fastparquet + datetime.today(), f"{table_name}/%Y/%m/%d/{table_name}_%H:%M:%S.parquet" + ) + parquet_file = df.to_parquet( + f"{table_name}.parquet", engine="pyarrow" + ) # or fastparquet client.upload_file(parquet_file, bucket, s3_key) - status['uploaded'].append(table_name) - + status["uploaded"].append(table_name) return status - def retrieve_secrets(): secret_name = "bentley-secrets" region_name = "eu-west-2" @@ -175,19 +183,23 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")): table_dfs[table] = pd.concat(list_of_df) return table_dfs + def bucket_name(bucket_prefix, client=boto3.client("s3")): response = client.list_buckets() bucket_filter = [ - bucket["Name"] for bucket in response["Buckets"] if bucket_prefix in bucket["Name"] + bucket["Name"] + for bucket in response["Buckets"] + if bucket_prefix in bucket["Name"] ] return bucket_filter[0] + def list_existing_s3_files(bucket_name, client=boto3.client("s3")): logging.info("Listing existing S3 files") try: - response = client.list_objects_v2(Bucket=bucket_name) + response = client.list_objects_v2(Bucket=bucket_name) if "Contents" in response: existing_files = [obj["Key"] for obj in response["Contents"]] @@ -198,4 +210,4 @@ def list_existing_s3_files(bucket_name, client=boto3.client("s3")): except ClientError as e: logger.error(f"Error listing S3 objects: {e}") - return existing_files \ No newline at end of file + return existing_files diff --git a/tests/test_fact_sales_order.py b/tests/test_fact_sales_order.py index 82845d7..87e3ade 100644 --- a/tests/test_fact_sales_order.py +++ b/tests/test_fact_sales_order.py @@ -1,57 +1,109 @@ -from src.fact_sales_order import create_dim_design, create_dim_staff, create_dim_currency +from src.fact_sales_order import ( + create_dim_design, + create_dim_staff, + create_dim_currency, +) import pandas as pd + class TestCreateDimDesign: def test_dim_design_returns_dataframe(self): - d = {"test": ["Hello", "Bye"], "design_id": ["Hello", "Bye"], "design_name": ["Hello", "Bye"], - "file_name": ["Hello", "Bye"], "file_location": ["Hello", "Bye"], "Hello": ["Hello", "Bye"]} + d = { + "test": ["Hello", "Bye"], + "design_id": ["Hello", "Bye"], + "design_name": ["Hello", "Bye"], + "file_name": ["Hello", "Bye"], + "file_location": ["Hello", "Bye"], + "Hello": ["Hello", "Bye"], + } test_df = {"design": pd.DataFrame(data=d)} result = create_dim_design(test_df) assert isinstance(result, pd.DataFrame) def test_dim_design_returns_correct_columns_and_values(self): - d = {"test": ["Hello", "Bye"], "design_id": ["Hello", "Bye"], "design_name": ["Hello", "Bye"], - "file_name": ["Hello", "Bye"], "file_location": ["Hello", "Bye"], "Hello": ["Hello", "Bye"]} + d = { + "test": ["Hello", "Bye"], + "design_id": ["Hello", "Bye"], + "design_name": ["Hello", "Bye"], + "file_name": ["Hello", "Bye"], + "file_location": ["Hello", "Bye"], + "Hello": ["Hello", "Bye"], + } test_df = {"design": pd.DataFrame(data=d)} result = create_dim_design(test_df) - d2 = {"design_id": ["Hello", "Bye"], "design_name": ["Hello", "Bye"], "file_name": ["Hello", "Bye"], - "file_location": ["Hello", "Bye"]} + d2 = { + "design_id": ["Hello", "Bye"], + "design_name": ["Hello", "Bye"], + "file_name": ["Hello", "Bye"], + "file_location": ["Hello", "Bye"], + } expected_df = pd.DataFrame(data=d2) expected_result = expected_df.copy() assert result.equals(expected_result) + class TestCreateDimStaff: def test_dim_staff_returns_dataframe(self): - d = {"staff_id": ["Hello", "Bye"], "first_name": ["Hello", "Bye"], "last_name": ["Hello", "Bye"], "department_id": ["Hello", "Bye"]} - d2 = {"department_name": ["Hello", "Bye"], "location": ["Hello", "Bye"], "email_address": ["Hello", "Bye"], "department_id": ["Hello", "Bye"]} + d = { + "staff_id": ["Hello", "Bye"], + "first_name": ["Hello", "Bye"], + "last_name": ["Hello", "Bye"], + "department_id": ["Hello", "Bye"], + } + d2 = { + "department_name": ["Hello", "Bye"], + "location": ["Hello", "Bye"], + "email_address": ["Hello", "Bye"], + "department_id": ["Hello", "Bye"], + } test_df = {"staff": pd.DataFrame(data=d), "department": pd.DataFrame(data=d2)} result = create_dim_staff(test_df) - assert isinstance(result, pd.DataFrame) + assert isinstance(result, pd.DataFrame) def test_dim_staff_returns_correct_columns_and_values(self): - d = {"staff_id": ["Hello", "Bye"], "first_name": ["Hello", "Bye"], "last_name": ["Hello", "Bye"], "department_id": ["Hello", "Bye"]} - d2 = {"department_name": ["Hello", "Bye"], "location": ["Hello", "Bye"], "email_address": ["Hello", "Bye"], "department_id": ["Hello", "Bye"]} + d = { + "staff_id": ["Hello", "Bye"], + "first_name": ["Hello", "Bye"], + "last_name": ["Hello", "Bye"], + "department_id": ["Hello", "Bye"], + } + d2 = { + "department_name": ["Hello", "Bye"], + "location": ["Hello", "Bye"], + "email_address": ["Hello", "Bye"], + "department_id": ["Hello", "Bye"], + } test_df = {"staff": pd.DataFrame(data=d), "department": pd.DataFrame(data=d2)} result = create_dim_staff(test_df) - expected_d = {"staff_id": ["Hello", "Bye"], "first_name": ["Hello", "Bye"], "last_name": ["Hello", "Bye"], "department_name": ["Hello", "Bye"], "location": ["Hello", "Bye"], "email_address": ["Hello", "Bye"]} + expected_d = { + "staff_id": ["Hello", "Bye"], + "first_name": ["Hello", "Bye"], + "last_name": ["Hello", "Bye"], + "department_name": ["Hello", "Bye"], + "location": ["Hello", "Bye"], + "email_address": ["Hello", "Bye"], + } expected_df = pd.DataFrame(data=expected_d) expected_result = expected_df.copy() - assert result.equals(expected_result) + assert result.equals(expected_result) + class TestCreateDimCurrency: def test_dim_currency_returns_dataframe(self): d = {"currency_id": [1, 2, 3], "currency_code": ["USD", "EUR", "GBP"]} test_df = {"currency": pd.DataFrame(data=d)} result = create_dim_currency(test_df) - assert isinstance(result, pd.DataFrame) - + assert isinstance(result, pd.DataFrame) + def test_dim_currency_returns_columns_and_values(self): d = {"currency_id": [1, 2, 3], "currency_code": ["USD", "EUR", "GBP"]} test_df = {"currency": pd.DataFrame(data=d)} result = create_dim_currency(test_df) - expected_d = {"currency_id": [1, 2, 3], "currency_code": ["USD", "EUR", "GBP"], "currency_name": ["US Dollar", "Euro", "Pound"]} + expected_d = { + "currency_id": [1, 2, 3], + "currency_code": ["USD", "EUR", "GBP"], + "currency_name": ["US Dollar", "Euro", "Pound"], + } expected_df = pd.DataFrame(data=expected_d) expected_result = expected_df.copy() assert result.equals(expected_result) - - \ No newline at end of file diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 516f83b..a91da92 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -39,7 +39,12 @@ class TestReadFromS3: ) print(result) expected_df = pd.DataFrame( - np.array([["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"]]), + np.array( + [ + ["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], + ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"], + ] + ), columns=["Food_type", "Flavour", "Colour", "last_updated"], ) assert isinstance(result, dict) @@ -56,7 +61,12 @@ class TestReadFromS3: tables, bucket="dummy_buc", client=s3_client ) expected_foods_df = pd.DataFrame( - np.array([["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"]]), + np.array( + [ + ["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], + ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"], + ] + ), columns=["Food_type", "Flavour", "Colour", "last_updated"], ) expected_cars_df = pd.DataFrame( @@ -72,5 +82,3 @@ class TestReadFromS3: assert list(result.keys()) == tables assert result["Foods"].eq(expected_foods_df, axis="columns").all(axis=None) assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None) - - -- cgit v1.2.3 From a69fe58b47bcc5ad02986bcf404f060774aec9a7 Mon Sep 17 00:00:00 2001 From: lian-manonog Date: Fri, 23 Aug 2024 16:22:52 +0100 Subject: wip: pushing again --- src/dataframes.py | 12 ++++++------ src/transform_lambda.py | 1 + tests/test_transform_lambda.py | 43 +++++++++++++++++++++++++++++++++++++++--- 3 files changed, 47 insertions(+), 9 deletions(-) (limited to 'tests/test_transform_lambda.py') diff --git a/src/dataframes.py b/src/dataframes.py index 684f102..18e1fac 100644 --- a/src/dataframes.py +++ b/src/dataframes.py @@ -1,11 +1,11 @@ import pandas as pd from bs4 import BeautifulSoup -from src.transform_lambda import read_from_s3_subfolder_to_df, tables -from src.extract_lambda import extract_bucket -import json -import boto3 -import re -from datetime import datetime as dt +# from src.transform_lambda import read_from_s3_subfolder_to_df, tables +# from src.extract_lambda import extract_bucket +# import json +# import boto3 +# import re +# from datetime import datetime as dt import requests # Table names: diff --git a/src/transform_lambda.py b/src/transform_lambda.py index defa15d..7677f66 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -207,5 +207,6 @@ def list_existing_s3_files(bucket_name, client=boto3.client("s3")): except ClientError as e: logger.error(f"Error listing S3 objects: {e}") + raise e return existing_files diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 37ca08f..06235f7 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -1,12 +1,19 @@ -from src.transform_lambda import read_from_s3_subfolder_to_df +from src.transform_lambda import read_from_s3_subfolder_to_df, list_existing_s3_files from moto import mock_aws import pytest import pandas as pd import os import boto3 +from botocore.exceptions import ClientError import numpy as np +# import caplog +import logging + +logger = logging.getLogger() +logger.setLevel(logging.INFO) + @pytest.fixture(scope="class") def aws_credentials(): os.environ["AWS_ACCESS_KEY_ID"] = "testing" @@ -23,7 +30,7 @@ def s3_client(aws_credentials): class TestReadFromS3: - @pytest.mark.skip(reason="The test is broken!") + # @pytest.mark.skip(reason="The test is broken!") def test_returns_dictionary_with_correct_value_pair(self, s3_client): s3_client.create_bucket( Bucket="dummy_buc", @@ -53,7 +60,7 @@ class TestReadFromS3: assert isinstance(result["Foods"], pd.DataFrame) assert result["Foods"].eq(expected_df, axis="columns").all(axis=None) - @pytest.mark.skip(reason="The test is broken!") + # @pytest.mark.skip(reason="The test is broken!") def test_returns_dictionary_of_dataframes_for_multiple_tables(self, s3_client): s3_client.upload_file( "tests/dummy_2.csv", "dummy_buc", "Cars/2024/08/21/Cars_14:03:56.csv" @@ -84,3 +91,33 @@ class TestReadFromS3: assert list(result.keys()) == tables assert result["Foods"].eq(expected_foods_df, axis="columns").all(axis=None) assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None) + +class TestListExistingFiles: + def test_functions_receives_error_if_no_bucket(self, s3_client, caplog): + caplog.set_level(logging.INFO) + + with pytest.raises(ClientError): + list_existing_s3_files('rando_bucket', client=s3_client) + + assert "Error listing S3 objects: An error occurred (NoSuchBucket) when calling the ListObjectsV2 operation: The specified bucket does not exist" in caplog.text + + def test_recieves_logger_error_if_no_files_listed(self, s3_client, caplog): + caplog.set_level(logging.INFO) + + s3_client.create_bucket( + Bucket='mock_bucket', + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"} + ) + response = list_existing_s3_files('mock_bucket', client=s3_client) + assert 'The bucket is empty' in caplog.text + + def test_retrieves_existing_files(self, s3_client, caplog): + caplog.set_level(logging.INFO) + + s3_client.upload_file( + "tests/dummy.txt", 'mock_bucket', "dummy.txt" + ) + result = list_existing_s3_files('mock_bucket', client=s3_client) + assert result == ["dummy.txt"] + + \ No newline at end of file -- cgit v1.2.3 From f1e10e1a2f573c152b19a630577a71ce9aff2bb4 Mon Sep 17 00:00:00 2001 From: lian-manonog Date: Fri, 23 Aug 2024 16:35:55 +0100 Subject: wip: writing more tests for the helper functions --- tests/test_transform_lambda.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) (limited to 'tests/test_transform_lambda.py') diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 06235f7..00f3d83 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -1,4 +1,4 @@ -from src.transform_lambda import read_from_s3_subfolder_to_df, list_existing_s3_files +from src.transform_lambda import read_from_s3_subfolder_to_df, list_existing_s3_files, bucket_name from moto import mock_aws import pytest import pandas as pd @@ -120,4 +120,14 @@ class TestListExistingFiles: result = list_existing_s3_files('mock_bucket', client=s3_client) assert result == ["dummy.txt"] - \ No newline at end of file +class TestBucketName: + def test_functions_retrieves_bucket(self, s3_client): + s3_client.create_bucket( + Bucket='mock_bucket', + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"} + ) + + bucket = bucket_name('mock_bucket', s3_client) + assert bucket == 'mock_bucket' + + # def test_ \ No newline at end of file -- cgit v1.2.3 From e51e9fc3c7fa886fe5e753bd123d45c8871673bc Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Tue, 27 Aug 2024 09:46:39 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in c68f63f according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/97 --- src/dataframes.py | 74 ++++++++++++++++++++---------------------- src/transform_lambda.py | 6 ++-- tests/test_transform_lambda.py | 44 +++++++++++++++---------- 3 files changed, 65 insertions(+), 59 deletions(-) (limited to 'tests/test_transform_lambda.py') diff --git a/src/dataframes.py b/src/dataframes.py index 94eb509..ab53063 100644 --- a/src/dataframes.py +++ b/src/dataframes.py @@ -21,10 +21,8 @@ def create_fact_sales_order(dict_of_df): df_sales.index.name = "sales_record_id" df_sales["created_date"] = pd.to_datetime(df_sales["created_at"]).dt.date df_sales["created_time"] = pd.to_datetime(df_sales["created_at"]).dt.time - df_sales["last_updated_date"] = pd.to_datetime( - df_sales["last_updated"]).dt.date - df_sales["last_updated_time"] = pd.to_datetime( - df_sales["last_updated"]).dt.time + df_sales["last_updated_date"] = pd.to_datetime(df_sales["last_updated"]).dt.date + df_sales["last_updated_time"] = pd.to_datetime(df_sales["last_updated"]).dt.time fact_sales_order = df_sales.loc[ :, [ @@ -76,7 +74,8 @@ def create_fact_payment(dict_of_df): df_payment["last_updated_date"] = df_payment["last_updated"].date() df_payment["last_updated_time"] = df_payment["last_updated"].time df_payment["payment_date"] = pd.to_datetime( - df_payment["payment_date"], format="%Y-%m-%d") + df_payment["payment_date"], format="%Y-%m-%d" + ) fact_payment = df_payment.loc[ :, [ @@ -113,16 +112,16 @@ def create_dim_location(dict_of_df): df_loc = ( dict_of_df["address"] .drop(labels=["created_at", "last_updated"], axis=1) - .rename(columns={"address_id": "location_id"})) + .rename(columns={"address_id": "location_id"}) + ) return df_loc - def create_dim_counterparty(dict_of_df): - df_prefixed_address=dict_of_df["address"].add_prefix( + df_prefixed_address = dict_of_df["address"].add_prefix( "counterparty_legal_", axis=1 ) - df_cp=pd.merge( + df_cp = pd.merge( dict_of_df["counterparty"], df_prefixed_address, left_on="legal_address_id", @@ -139,51 +138,51 @@ def create_dim_counterparty(dict_of_df): def create_dim_date(dict_of_df): - fact_dfs=[ + fact_dfs = [ create_fact_payment(dict_of_df), create_fact_purchase_orders(dict_of_df), create_fact_sales_order(dict_of_df), ] - date_col_names=[ + date_col_names = [ col_name for col_name in list(fact_dfs[0].columns) if "date" in col_name ] - list_of_date_columns=[] + list_of_date_columns = [] for df in fact_dfs: for col in date_col_names: list_of_date_columns.append(df[col]) - sr_date=pd.array(pd.concat(list_of_date_columns), dtype="datetime64[ns]") - df_date=pd.DataFrame(data=sr_date, columns=["date_id"]) + 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["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 - df_date["day_of_week"]=df_date["date_id"].dt.dayofweek - df_date["day_name"]=df_date["date_id"].dt.day_name() - df_date["month_name"]=df_date["date_id"].dt.month_name() - df_date["quarter"]=df_date["date_id"].dt.quarter + 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 + df_date["day_of_week"] = df_date["date_id"].dt.dayofweek + df_date["day_name"] = df_date["date_id"].dt.day_name() + df_date["month_name"] = df_date["date_id"].dt.month_name() + df_date["quarter"] = df_date["date_id"].dt.quarter return df_date # tests passed def scrape_currency_names(): - response=requests.get("https://www.xe.com/currency/").content - soup=BeautifulSoup(response, "html.parser") - currency=[ + response = requests.get("https://www.xe.com/currency/").content + soup = BeautifulSoup(response, "html.parser") + currency = [ item.text for item in soup.findAll("a", attrs={"class": "sc-299dec64-6 fZPTSw"}) ] - sr=pd.Series(currency) - df_cur=sr.str.split(pat=" - ", expand=True).rename( + sr = pd.Series(currency) + df_cur = sr.str.split(pat=" - ", expand=True).rename( {0: "currency_code", 1: "currency_name"}, axis=1 ) return df_cur + # tests passed 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 = 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" ) return dim_cur @@ -191,33 +190,32 @@ def create_dim_currency(dict_of_df, names=scrape_currency_names()): # tests passed + 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"]] + df_payment_type = dict_of_df["payment_type"] + dim_payment_type = df_payment_type.loc[:, ["payment_type_id", "payment_type_name"]] return dim_payment_type - # tests passed def create_dim_design(dict_of_df): - df_design=dict_of_df["design"] - dim_design=df_design.loc[ + df_design = dict_of_df["design"] + dim_design = df_design.loc[ :, ["design_id", "design_name", "file_name", "file_location"] ] return dim_design - # tests passed + def create_dim_staff(dict_of_df): - staff_department=pd.merge( + 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", diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 565b4ee..2cd9272 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -11,7 +11,6 @@ from pg8000.native import Connection, InterfaceError from datetime import datetime - class DBConnectionException(Exception): """Wraps pg8000.native Error or DatabaseError.""" @@ -212,5 +211,6 @@ def list_existing_s3_files(bucket_name, client=boto3.client("s3")): return existing_files -if __name__ == '__main__': - lambda_handler({}, '') \ No newline at end of file + +if __name__ == "__main__": + lambda_handler({}, "") diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 00f3d83..5ed743e 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -1,4 +1,8 @@ -from src.transform_lambda import read_from_s3_subfolder_to_df, list_existing_s3_files, bucket_name +from src.transform_lambda import ( + read_from_s3_subfolder_to_df, + list_existing_s3_files, + bucket_name, +) from moto import mock_aws import pytest import pandas as pd @@ -6,14 +10,15 @@ import os import boto3 from botocore.exceptions import ClientError import numpy as np + # import caplog import logging - logger = logging.getLogger() logger.setLevel(logging.INFO) + @pytest.fixture(scope="class") def aws_credentials(): os.environ["AWS_ACCESS_KEY_ID"] = "testing" @@ -92,42 +97,45 @@ class TestReadFromS3: assert result["Foods"].eq(expected_foods_df, axis="columns").all(axis=None) assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None) + class TestListExistingFiles: def test_functions_receives_error_if_no_bucket(self, s3_client, caplog): caplog.set_level(logging.INFO) with pytest.raises(ClientError): - list_existing_s3_files('rando_bucket', client=s3_client) + list_existing_s3_files("rando_bucket", client=s3_client) - assert "Error listing S3 objects: An error occurred (NoSuchBucket) when calling the ListObjectsV2 operation: The specified bucket does not exist" in caplog.text + assert ( + "Error listing S3 objects: An error occurred (NoSuchBucket) when calling the ListObjectsV2 operation: The specified bucket does not exist" + in caplog.text + ) def test_recieves_logger_error_if_no_files_listed(self, s3_client, caplog): caplog.set_level(logging.INFO) s3_client.create_bucket( - Bucket='mock_bucket', - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"} + Bucket="mock_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, ) - response = list_existing_s3_files('mock_bucket', client=s3_client) - assert 'The bucket is empty' in caplog.text + response = list_existing_s3_files("mock_bucket", client=s3_client) + assert "The bucket is empty" in caplog.text def test_retrieves_existing_files(self, s3_client, caplog): caplog.set_level(logging.INFO) - s3_client.upload_file( - "tests/dummy.txt", 'mock_bucket', "dummy.txt" - ) - result = list_existing_s3_files('mock_bucket', client=s3_client) + s3_client.upload_file("tests/dummy.txt", "mock_bucket", "dummy.txt") + result = list_existing_s3_files("mock_bucket", client=s3_client) assert result == ["dummy.txt"] + class TestBucketName: def test_functions_retrieves_bucket(self, s3_client): s3_client.create_bucket( - Bucket='mock_bucket', - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"} + Bucket="mock_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, ) - - bucket = bucket_name('mock_bucket', s3_client) - assert bucket == 'mock_bucket' - # def test_ \ No newline at end of file + bucket = bucket_name("mock_bucket", s3_client) + assert bucket == "mock_bucket" + + # def test_ -- cgit v1.2.3 From 836f71dbea59a35b2eeeeeb982a73c4366089722 Mon Sep 17 00:00:00 2001 From: HastarTara Date: Tue, 27 Aug 2024 12:33:03 +0100 Subject: tests for bucket_name helper --- src/transform_lambda.py | 17 +++++++++----- tests/test_transform_lambda.py | 52 +++++++++++++++++++++++++++--------------- 2 files changed, 44 insertions(+), 25 deletions(-) (limited to 'tests/test_transform_lambda.py') diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 2cd9272..cd9541d 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -1,3 +1,4 @@ +from src.dataframes import * import json import boto3 import re @@ -5,7 +6,6 @@ import logging import pandas as pd import pyarrow as pa import pyarrow.parquet as pq -from dataframes import * from botocore.exceptions import ClientError from pg8000.native import Connection, InterfaceError from datetime import datetime @@ -183,13 +183,18 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")): def bucket_name(bucket_prefix, client=boto3.client("s3")): + # response = client.list_buckets() + # for bucket in response["Buckets"]: + # if bucket_prefix in bucket["Name"]: + # return bucket["Name"] + + response = client.list_buckets() bucket_filter = [ - bucket["Name"] - for bucket in response["Buckets"] - if bucket_prefix in bucket["Name"] - ] - + bucket["Name"] + for bucket in response["Buckets"] + if bucket_prefix in bucket["Name"] + ] return bucket_filter[0] diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 5ed743e..cc4e07a 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -33,22 +33,36 @@ def s3_client(aws_credentials): with mock_aws(): yield boto3.client("s3") +@pytest.fixture(scope="class") +def mock_extract_bucket(s3_client): + mock_extract_bucket = s3_client.create_bucket( + Bucket="dummy_extract_buc", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + return mock_extract_bucket + +@pytest.fixture(scope="class") +def mock_transform_bucket(s3_client): + mock_transform_bucket = s3_client.create_bucket( + Bucket="dummy_transform_buc", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + return mock_transform_bucket + + class TestReadFromS3: # @pytest.mark.skip(reason="The test is broken!") - def test_returns_dictionary_with_correct_value_pair(self, s3_client): - s3_client.create_bucket( - Bucket="dummy_buc", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) + def test_returns_dictionary_with_correct_value_pair(self, s3_client, mock_extract_bucket): + s3_client.upload_file( "tests/dummy_identical.csv", - "dummy_buc", + "dummy_extract_buc", "Foods/2024/08/21/Foods_12:03:10.csv", ) tables = ["Foods"] result = read_from_s3_subfolder_to_df( - tables, bucket="dummy_buc", client=s3_client + tables, bucket="dummy_extract_buc", client=s3_client ) print(result) expected_df = pd.DataFrame( @@ -66,13 +80,13 @@ class TestReadFromS3: assert result["Foods"].eq(expected_df, axis="columns").all(axis=None) # @pytest.mark.skip(reason="The test is broken!") - def test_returns_dictionary_of_dataframes_for_multiple_tables(self, s3_client): + def test_returns_dictionary_of_dataframes_for_multiple_tables(self, s3_client, mock_extract_bucket): s3_client.upload_file( - "tests/dummy_2.csv", "dummy_buc", "Cars/2024/08/21/Cars_14:03:56.csv" + "tests/dummy_2.csv", "dummy_extract_buc", "Cars/2024/08/21/Cars_14:03:56.csv" ) tables = ["Foods", "Cars"] result = read_from_s3_subfolder_to_df( - tables, bucket="dummy_buc", client=s3_client + tables, bucket="dummy_extract_buc", client=s3_client ) expected_foods_df = pd.DataFrame( np.array( @@ -95,7 +109,7 @@ class TestReadFromS3: ) assert list(result.keys()) == tables assert result["Foods"].eq(expected_foods_df, axis="columns").all(axis=None) - assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None) + # assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None) class TestListExistingFiles: @@ -129,13 +143,13 @@ class TestListExistingFiles: class TestBucketName: - def test_functions_retrieves_bucket(self, s3_client): - s3_client.create_bucket( - Bucket="mock_bucket", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) + def test_functions_retrieves__extractbucket(self, mock_extract_bucket, mock_transform_bucket,s3_client): + + bucket = bucket_name("dummy_extract_buc", s3_client) + assert bucket == "dummy_extract_buc" - bucket = bucket_name("mock_bucket", s3_client) - assert bucket == "mock_bucket" - # def test_ + def test_transform_bucket_name(self, mock_extract_bucket, mock_transform_bucket, s3_client): + bucket2 = bucket_name('dummy_transform_buc', s3_client) + assert bucket2 == 'dummy_transform_buc' + \ No newline at end of file -- cgit v1.2.3 From ad357ff34202827720dc216562dfbb0fbd65c297 Mon Sep 17 00:00:00 2001 From: HastarTara Date: Tue, 27 Aug 2024 17:02:25 +0100 Subject: test updates to transform lambda handler --- car_data.parquet | Bin 0 -> 2827 bytes src/transform_lambda.py | 59 ++++++++++++++++++++++++----------------- tests/test_transform_lambda.py | 39 +++++++++++++++++++++++++-- 3 files changed, 71 insertions(+), 27 deletions(-) create mode 100644 car_data.parquet (limited to 'tests/test_transform_lambda.py') diff --git a/car_data.parquet b/car_data.parquet new file mode 100644 index 0000000..1853af6 Binary files /dev/null and b/car_data.parquet differ diff --git a/src/transform_lambda.py b/src/transform_lambda.py index cd9541d..9830e0f 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -9,7 +9,7 @@ import pyarrow.parquet as pq from botocore.exceptions import ClientError from pg8000.native import Connection, InterfaceError from datetime import datetime - +import io class DBConnectionException(Exception): """Wraps pg8000.native Error or DatabaseError.""" @@ -59,6 +59,8 @@ def lambda_handler(event, context): TABLES, bucket_name("extract"), client=boto3.client("s3") ) + print(dict_of_df) + immutable_df_dict = { "dim_counterparty": create_dim_counterparty(dict_of_df), "dim_date": create_dim_date(dict_of_df), @@ -106,7 +108,7 @@ def process_to_parquet_and_upload_to_s3( immutable_df_dict, mutable_df_dict, bucket, - client=boto3.client("s3"), + client=boto3.client("s3") ): status = {"uploaded": [], "not_uploaded": []} @@ -114,21 +116,25 @@ def process_to_parquet_and_upload_to_s3( if table_name in existing_s3_files: status["not_uploaded"].append(table_name) else: - parquet_file = df.to_parquet( - f"{table_name}.parquet", engine="pyarrow" - ) # or fastparquet - client.upload_file(parquet_file, bucket, f"{table_name}.parquet") + parquet_buffer = io.BytesIO() + + df.to_parquet(parquet_buffer, engine="pyarrow") # or engine="fastparquet" + + parquet_buffer.seek(0) + + client.upload_fileobj(parquet_buffer, bucket, f"{table_name}.parquet") + status["uploaded"].append(table_name) - for table_name, df in mutable_df_dict.items(): - s3_key = datetime.strftime( - datetime.today(), f"{table_name}/%Y/%m/%d/{table_name}_%H:%M:%S.parquet" - ) - parquet_file = df.to_parquet( - f"{table_name}.parquet", engine="pyarrow" - ) # or fastparquet - client.upload_file(parquet_file, bucket, s3_key) - status["uploaded"].append(table_name) + # for table_name, df in mutable_df_dict.items(): + # s3_key = datetime.strftime( + # datetime.today(), f"{table_name}/%Y/%m/%d/{table_name}_%H:%M:%S.parquet" + # ) + # parquet_file = df.to_parquet( + # f"{table_name}.parquet", engine="pyarrow" + # ) # or fastparquet + # client.upload_file(parquet_file, bucket, s3_key) + # status["uploaded"].append(table_name) return status @@ -182,20 +188,23 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")): return table_dfs + + def bucket_name(bucket_prefix, client=boto3.client("s3")): - # response = client.list_buckets() - # for bucket in response["Buckets"]: - # if bucket_prefix in bucket["Name"]: - # return bucket["Name"] - - - response = client.list_buckets() - bucket_filter = [ + + response = client.list_buckets() + bucket_filter = [ bucket["Name"] for bucket in response["Buckets"] if bucket_prefix in bucket["Name"] - ] - return bucket_filter[0] + ] + if not bucket_filter: + raise ValueError(f"No bucket found with prefix: {bucket_prefix}") + + return bucket_filter[0] + + + def list_existing_s3_files(bucket_name, client=boto3.client("s3")): diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index cc4e07a..b4836c2 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -1,7 +1,7 @@ from src.transform_lambda import ( read_from_s3_subfolder_to_df, list_existing_s3_files, - bucket_name, + bucket_name, process_to_parquet_and_upload_to_s3 ) from moto import mock_aws import pytest @@ -152,4 +152,39 @@ class TestBucketName: def test_transform_bucket_name(self, mock_extract_bucket, mock_transform_bucket, s3_client): bucket2 = bucket_name('dummy_transform_buc', s3_client) assert bucket2 == 'dummy_transform_buc' - \ No newline at end of file + + + def test_recieves_error_when_bucket_doesnt_exist(self, mock_extract_bucket, s3_client): + s3_client.delete_bucket(Bucket='dummy_extract_buc') + with pytest.raises(ValueError): + bucket_name('dummy_extract_buc', s3_client) + + + + + + +class TestProcessToParquetUploadS3: + def test_func_uploads_to_s3(self, mock_transform_bucket, s3_client): + + expected_cars_df = pd.DataFrame( + np.array( + [ + ["Truck", "Chevrolet", "Grey"], + ["Convertible", "Mercedes", "Red"], + ["Van", "Volkswagen", "Blue"], + ] + ), + columns=["Car_type", "Brand", "Colour"], + ) + mock_dim_dict = {'car_data': expected_cars_df} + + response = process_to_parquet_and_upload_to_s3([], mock_dim_dict, {}, mock_transform_bucket, s3_client) + + + assert response == {"uploaded": ["car_data"], "not_uploaded": []} + + + + + -- cgit v1.2.3 From 3f24ec753902feecec4c17e2877e19853bde1bb2 Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Wed, 28 Aug 2024 09:59:43 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in ad357ff according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/105 --- src/transform_lambda.py | 40 +++++++++++------------ tests/test_transform_lambda.py | 73 +++++++++++++++++++++--------------------- 2 files changed, 55 insertions(+), 58 deletions(-) (limited to 'tests/test_transform_lambda.py') diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 9830e0f..3b1e9e6 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -11,6 +11,7 @@ from pg8000.native import Connection, InterfaceError from datetime import datetime import io + class DBConnectionException(Exception): """Wraps pg8000.native Error or DatabaseError.""" @@ -108,7 +109,7 @@ def process_to_parquet_and_upload_to_s3( immutable_df_dict, mutable_df_dict, bucket, - client=boto3.client("s3") + client=boto3.client("s3"), ): status = {"uploaded": [], "not_uploaded": []} @@ -117,13 +118,14 @@ def process_to_parquet_and_upload_to_s3( status["not_uploaded"].append(table_name) else: parquet_buffer = io.BytesIO() - - df.to_parquet(parquet_buffer, engine="pyarrow") # or engine="fastparquet" - + + # or engine="fastparquet" + df.to_parquet(parquet_buffer, engine="pyarrow") + parquet_buffer.seek(0) - + client.upload_fileobj(parquet_buffer, bucket, f"{table_name}.parquet") - + status["uploaded"].append(table_name) # for table_name, df in mutable_df_dict.items(): @@ -188,23 +190,17 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")): return table_dfs - - def bucket_name(bucket_prefix, client=boto3.client("s3")): - - response = client.list_buckets() - bucket_filter = [ - bucket["Name"] - for bucket in response["Buckets"] - if bucket_prefix in bucket["Name"] - ] - if not bucket_filter: - raise ValueError(f"No bucket found with prefix: {bucket_prefix}") - - return bucket_filter[0] - - - + response = client.list_buckets() + bucket_filter = [ + bucket["Name"] + for bucket in response["Buckets"] + if bucket_prefix in bucket["Name"] + ] + if not bucket_filter: + raise ValueError(f"No bucket found with prefix: {bucket_prefix}") + + return bucket_filter[0] def list_existing_s3_files(bucket_name, client=boto3.client("s3")): diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index b4836c2..6cf3a09 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -1,7 +1,8 @@ from src.transform_lambda import ( read_from_s3_subfolder_to_df, list_existing_s3_files, - bucket_name, process_to_parquet_and_upload_to_s3 + bucket_name, + process_to_parquet_and_upload_to_s3, ) from moto import mock_aws import pytest @@ -33,28 +34,30 @@ def s3_client(aws_credentials): with mock_aws(): yield boto3.client("s3") + @pytest.fixture(scope="class") def mock_extract_bucket(s3_client): mock_extract_bucket = s3_client.create_bucket( - Bucket="dummy_extract_buc", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) + Bucket="dummy_extract_buc", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) return mock_extract_bucket - + + @pytest.fixture(scope="class") def mock_transform_bucket(s3_client): mock_transform_bucket = s3_client.create_bucket( - Bucket="dummy_transform_buc", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) + Bucket="dummy_transform_buc", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) return mock_transform_bucket - class TestReadFromS3: # @pytest.mark.skip(reason="The test is broken!") - def test_returns_dictionary_with_correct_value_pair(self, s3_client, mock_extract_bucket): - + def test_returns_dictionary_with_correct_value_pair( + self, s3_client, mock_extract_bucket + ): s3_client.upload_file( "tests/dummy_identical.csv", "dummy_extract_buc", @@ -80,9 +83,13 @@ class TestReadFromS3: assert result["Foods"].eq(expected_df, axis="columns").all(axis=None) # @pytest.mark.skip(reason="The test is broken!") - def test_returns_dictionary_of_dataframes_for_multiple_tables(self, s3_client, mock_extract_bucket): + def test_returns_dictionary_of_dataframes_for_multiple_tables( + self, s3_client, mock_extract_bucket + ): s3_client.upload_file( - "tests/dummy_2.csv", "dummy_extract_buc", "Cars/2024/08/21/Cars_14:03:56.csv" + "tests/dummy_2.csv", + "dummy_extract_buc", + "Cars/2024/08/21/Cars_14:03:56.csv", ) tables = ["Foods", "Cars"] result = read_from_s3_subfolder_to_df( @@ -143,30 +150,28 @@ class TestListExistingFiles: class TestBucketName: - def test_functions_retrieves__extractbucket(self, mock_extract_bucket, mock_transform_bucket,s3_client): - + def test_functions_retrieves__extractbucket( + self, mock_extract_bucket, mock_transform_bucket, s3_client + ): bucket = bucket_name("dummy_extract_buc", s3_client) assert bucket == "dummy_extract_buc" + def test_transform_bucket_name( + self, mock_extract_bucket, mock_transform_bucket, s3_client + ): + bucket2 = bucket_name("dummy_transform_buc", s3_client) + assert bucket2 == "dummy_transform_buc" - def test_transform_bucket_name(self, mock_extract_bucket, mock_transform_bucket, s3_client): - bucket2 = bucket_name('dummy_transform_buc', s3_client) - assert bucket2 == 'dummy_transform_buc' - - - def test_recieves_error_when_bucket_doesnt_exist(self, mock_extract_bucket, s3_client): - s3_client.delete_bucket(Bucket='dummy_extract_buc') + def test_recieves_error_when_bucket_doesnt_exist( + self, mock_extract_bucket, s3_client + ): + s3_client.delete_bucket(Bucket="dummy_extract_buc") with pytest.raises(ValueError): - bucket_name('dummy_extract_buc', s3_client) - - - - + bucket_name("dummy_extract_buc", s3_client) class TestProcessToParquetUploadS3: def test_func_uploads_to_s3(self, mock_transform_bucket, s3_client): - expected_cars_df = pd.DataFrame( np.array( [ @@ -177,14 +182,10 @@ class TestProcessToParquetUploadS3: ), columns=["Car_type", "Brand", "Colour"], ) - mock_dim_dict = {'car_data': expected_cars_df} - - response = process_to_parquet_and_upload_to_s3([], mock_dim_dict, {}, mock_transform_bucket, s3_client) + mock_dim_dict = {"car_data": expected_cars_df} + response = process_to_parquet_and_upload_to_s3( + [], mock_dim_dict, {}, mock_transform_bucket, s3_client + ) assert response == {"uploaded": ["car_data"], "not_uploaded": []} - - - - - -- cgit v1.2.3 From 6235a2bb04b60d57a41196b07bbf0296920c6980 Mon Sep 17 00:00:00 2001 From: T-Aji 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 'tests/test_transform_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, "<<<<<<