From 1a145a36d524a785c821aafbdb3512c24be6c57e Mon Sep 17 00:00:00 2001 From: Ang Bel Date: Tue, 27 Aug 2024 17:00:04 +0100 Subject: test: transform refactoring - it now loads parquet files into s3 bucket --- src/transform_lambda.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'src/transform_lambda.py') diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 2cd9272..ccf90e5 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -117,7 +117,7 @@ def process_to_parquet_and_upload_to_s3( parquet_file = df.to_parquet( f"{table_name}.parquet", engine="pyarrow" ) # or fastparquet - client.upload_file(parquet_file, bucket, f"{table_name}.parquet") + client.upload_file(f"{table_name}.parquet", bucket, f"{table_name}.parquet") #changed parquet_file variable to the file name status["uploaded"].append(table_name) for table_name, df in mutable_df_dict.items(): @@ -127,7 +127,7 @@ def process_to_parquet_and_upload_to_s3( parquet_file = df.to_parquet( f"{table_name}.parquet", engine="pyarrow" ) # or fastparquet - client.upload_file(parquet_file, bucket, s3_key) + client.upload_file(f"{table_name}.parquet", bucket, s3_key) status["uploaded"].append(table_name) return status @@ -203,7 +203,7 @@ def list_existing_s3_files(bucket_name, client=boto3.client("s3")): existing_files = [obj["Key"] for obj in response["Contents"]] else: logger.error("The bucket is empty") - return None + return [] #changed from None to [] so it is an iterable except ClientError as e: logger.error(f"Error listing S3 objects: {e}") -- cgit v1.2.3 From dc095acd4d5b9f73a716a076ce601c3810f9635b Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Tue, 27 Aug 2024 16:01:11 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 1a145a3 according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/101 --- src/dataframes.py | 236 ++++++++++++++++++++++++++------------- src/transform_lambda.py | 5 +- tests/test_dataframes.py | 282 +++++++++++++++++++++++++++++++++++------------ 3 files changed, 375 insertions(+), 148 deletions(-) (limited to 'src/transform_lambda.py') diff --git a/src/dataframes.py b/src/dataframes.py index 9d0f2ac..f122368 100644 --- a/src/dataframes.py +++ b/src/dataframes.py @@ -2,7 +2,7 @@ import pandas as pd from bs4 import BeautifulSoup import requests -#Table names: +# Table names: # fact_sales_order # fact_purchase_orders # fact_payment @@ -16,133 +16,213 @@ import requests # dim_counterparty -#no test, same as fact_payment +# no test, same as fact_payment def create_fact_sales_order(dict_of_df): df_sales = dict_of_df["sales_order"] df_sales.index.name = "sales_record_id" - df_sales["created_date"] = 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["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 -#no test, same as fact_payment + +# no test, same as fact_payment + + 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 - df_po['last_updated_date'] = df_po['last_updated'].astype('datetime64[ns]').dt.date - df_po['last_updated_time'] = df_po['last_updated'].astype('datetime64[ns]').dt.floor('s').dt.time - df_po['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 = df_po.drop(labels=['created_at','last_updated'],axis=1) + 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 + ) + df_po["last_updated_date"] = df_po["last_updated"].astype("datetime64[ns]").dt.date + df_po["last_updated_time"] = ( + df_po["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time + ) + df_po["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 = df_po.drop(labels=["created_at", "last_updated"], axis=1) df_po.reset_index(inplace=True) return df_po -#test passed + +# test passed + + 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 - df_payment["created_time"] = df_payment["created_at"].astype('datetime64[ns]').dt.floor('s').dt.time - df_payment["last_updated_date"] = df_payment["last_updated"].astype('datetime64[ns]').dt.date - df_payment["last_updated_time"] = df_payment["last_updated"].astype('datetime64[ns]').dt.floor('s').dt.time - 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["created_date"] = ( + df_payment["created_at"].astype("datetime64[ns]").dt.date + ) + df_payment["created_time"] = ( + df_payment["created_at"].astype("datetime64[ns]").dt.floor("s").dt.time + ) + df_payment["last_updated_date"] = ( + df_payment["last_updated"].astype("datetime64[ns]").dt.date + ) + df_payment["last_updated_time"] = ( + df_payment["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time + ) + 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 -#test passed + +# test passed + + def create_dim_transaction(dict_of_df): - df_transaction = dict_of_df["transaction"].drop(labels=['created_at', 'last_updated'], axis=1) + df_transaction = dict_of_df["transaction"].drop( + labels=["created_at", "last_updated"], axis=1 + ) return df_transaction -#test passed + +# test passed + + 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'}) + df_loc = ( + dict_of_df["address"] + .drop(labels=["created_at", "last_updated"], axis=1) + .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('counterparty_legal_', axis=1) - df_cp = pd.merge(dict_of_df['counterparty'], - df_prefixed_address, - left_on="legal_address_id", - right_on="counterparty_legal_address_id", - how="outer") - df_cp.drop(columns=["legal_address_id","counterparty_legal_address_id"],inplace=True) + 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="counterparty_legal_address_id", + how="outer", + ) + df_cp.drop( + columns=["legal_address_id", "counterparty_legal_address_id"], inplace=True + ) return df_cp -#test passed + +# test passed + + def create_dim_date(dict_of_df): - fact_dfs = [create_fact_payment(dict_of_df), create_fact_purchase_orders(dict_of_df), create_fact_sales_order(dict_of_df)] + fact_dfs = [ + create_fact_payment(dict_of_df), + create_fact_purchase_orders(dict_of_df), + create_fact_sales_order(dict_of_df), + ] list_of_date_columns = [] for df in fact_dfs: - date_col_names = [col_name for col_name in list(df.columns) if '_date' in col_name] + date_col_names = [ + col_name for col_name in list(df.columns) if "_date" in col_name + ] for col in date_col_names: list_of_date_columns.append(df[col]) - 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 = [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) - 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,names,left_on='currency_code',right_on='currency_code',how='inner') - return dim_cur -#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"]] - return dim_payment_type +# tests passed -#tests passed -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"]] - return dim_design -#tests passed -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']] - return dim_staff +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"}) + ] + 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, names, left_on="currency_code", right_on="currency_code", how="inner" + ) + return dim_cur +# 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"]] + return dim_payment_type +# tests passed +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"] + ] + return dim_design +# tests passed +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", + ], + ] + return dim_staff diff --git a/src/transform_lambda.py b/src/transform_lambda.py index ccf90e5..93b2284 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -117,7 +117,8 @@ def process_to_parquet_and_upload_to_s3( parquet_file = df.to_parquet( f"{table_name}.parquet", engine="pyarrow" ) # or fastparquet - client.upload_file(f"{table_name}.parquet", bucket, f"{table_name}.parquet") #changed parquet_file variable to the file name + # changed parquet_file variable to the file name + client.upload_file(f"{table_name}.parquet", bucket, f"{table_name}.parquet") status["uploaded"].append(table_name) for table_name, df in mutable_df_dict.items(): @@ -203,7 +204,7 @@ def list_existing_s3_files(bucket_name, client=boto3.client("s3")): existing_files = [obj["Key"] for obj in response["Contents"]] else: logger.error("The bucket is empty") - return [] #changed from None to [] so it is an iterable + return [] # changed from None to [] so it is an iterable except ClientError as e: logger.error(f"Error listing S3 objects: {e}") diff --git a/tests/test_dataframes.py b/tests/test_dataframes.py index adbb5ed..c9ff43f 100644 --- a/tests/test_dataframes.py +++ b/tests/test_dataframes.py @@ -3,42 +3,88 @@ import pandas as pd from unittest.mock import patch from datetime import datetime as dt + 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 TestCreatePaymentType: def test_create_dim_payment_type_returns_correct_columns_and_values(self): @@ -46,100 +92,200 @@ class TestCreatePaymentType: test_df = {"payment_type": pd.DataFrame(data=d)} result = create_dim_payment_type(test_df) expected_columns = ["payment_type_id", "payment_type_name"] - expected_d = {"payment_type_id": ["Hello", "Bye"], "payment_type_name": ["Hello", "Bye"]} + expected_d = { + "payment_type_id": ["Hello", "Bye"], + "payment_type_name": ["Hello", "Bye"], + } expected_df = pd.DataFrame(data=expected_d) assert isinstance(result, pd.DataFrame) assert list(result.columns) == expected_columns assert result.equals(expected_df) + class TestCreateDimCounterparty: - def test_create_dim_counterparty_type_returns_correct_columns_and_object(self): - data_l = pd.DataFrame(data={"counterparty_id": ["Hello", "Bye"], - "counterparty_legal_name": ["Hello", "Bye"], - "commercial_contact": ["Hello", "Bye"], - "legal_address_id": ["bond street", "regent street"]}) - data_a = pd.DataFrame(data={"address_id":["bond street", "regent street"], - "postcode":[98365,93753]}) - test_df = {"address": data_a,"counterparty":data_l} + data_l = pd.DataFrame( + data={ + "counterparty_id": ["Hello", "Bye"], + "counterparty_legal_name": ["Hello", "Bye"], + "commercial_contact": ["Hello", "Bye"], + "legal_address_id": ["bond street", "regent street"], + } + ) + data_a = pd.DataFrame( + data={ + "address_id": ["bond street", "regent street"], + "postcode": [98365, 93753], + } + ) + test_df = {"address": data_a, "counterparty": data_l} result = create_dim_counterparty(test_df) - expected_columns = ["counterparty_id", "counterparty_legal_name", - "commercial_contact", "counterparty_legal_postcode"] + expected_columns = [ + "counterparty_id", + "counterparty_legal_name", + "commercial_contact", + "counterparty_legal_postcode", + ] print(data_l) print(data_a) assert isinstance(result, pd.DataFrame) assert list(result.columns) == expected_columns + class TestCreateDimCurrency: - def test_dim_currency_returns_columns_and_values(self): - nones = [None,None,None] - d = {"currency_id": [1, 2, 3], "currency_code": ["USD", "EUR", "GBP"],"created_at":nones,"last_updated":nones} + nones = [None, None, None] + d = { + "currency_id": [1, 2, 3], + "currency_code": ["USD", "EUR", "GBP"], + "created_at": nones, + "last_updated": nones, + } test_df = {"currency": pd.DataFrame(data=d)} - scraper_output = pd.DataFrame({"currency_code":["RUS","USD","PHP","GBP","EUR"],"currency_name":["Rubble","US Dollar","Peso","Pound","Euro"]}) - result = create_dim_currency(test_df,names=scraper_output).sort_values(by="currency_code",axis=0) - 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).sort_values(by="currency_code",axis=0) - assert isinstance(result, pd.DataFrame) - assert result.equals(expected_df) + scraper_output = pd.DataFrame( + { + "currency_code": ["RUS", "USD", "PHP", "GBP", "EUR"], + "currency_name": ["Rubble", "US Dollar", "Peso", "Pound", "Euro"], + } + ) + result = create_dim_currency(test_df, names=scraper_output).sort_values( + by="currency_code", axis=0 + ) + 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).sort_values( + by="currency_code", axis=0 + ) + assert isinstance(result, pd.DataFrame) + assert result.equals(expected_df) def test_scrape_currency_names_returns_dataframe_with_correct_collumns(self): result = scrape_currency_names() - assert isinstance(result,pd.DataFrame) - assert list(result.columns) == ['currency_code', 'currency_name'] + assert isinstance(result, pd.DataFrame) + assert list(result.columns) == ["currency_code", "currency_name"] -class TestCreateDimDate: +class TestCreateDimDate: def test_returns_required_columns(self): - df_one = pd.DataFrame(data={'updated_date':dt(2020, 5, 17),'created_date':dt(2021, 5, 13),'not_dat':None},index=[0]) - df_two = pd.DataFrame(data={'updated_date':dt(2020, 5, 17),'created_date':dt(2021, 9, 13)},index=[0]) - df_three = pd.DataFrame(data={'updated_date':dt(2022, 5, 17),'created_date':dt(2023, 5, 13)},index=[0]) - expected_df = pd.DataFrame(data= - [[dt(2020,5,17),2020,5,17,6,'Sunday','May',2], - [dt(2021,5,13),2021,5,13,3,'Thursday','May',2], - [dt(2021,9,13),2021,9,13,0,'Monday','September',3], - [dt(2022,5,17),2022,5,17,1,'Tuesday','May',2], - [dt(2023,5,13),2023,5,13,5,'Saturday','May',2]], - columns=['date_id','year','month','day','day_of_week','day_name','month_name','quarter']) + df_one = pd.DataFrame( + data={ + "updated_date": dt(2020, 5, 17), + "created_date": dt(2021, 5, 13), + "not_dat": None, + }, + index=[0], + ) + df_two = pd.DataFrame( + data={"updated_date": dt(2020, 5, 17), "created_date": dt(2021, 9, 13)}, + index=[0], + ) + df_three = pd.DataFrame( + data={"updated_date": dt(2022, 5, 17), "created_date": dt(2023, 5, 13)}, + index=[0], + ) + expected_df = pd.DataFrame( + data=[ + [dt(2020, 5, 17), 2020, 5, 17, 6, "Sunday", "May", 2], + [dt(2021, 5, 13), 2021, 5, 13, 3, "Thursday", "May", 2], + [dt(2021, 9, 13), 2021, 9, 13, 0, "Monday", "September", 3], + [dt(2022, 5, 17), 2022, 5, 17, 1, "Tuesday", "May", 2], + [dt(2023, 5, 13), 2023, 5, 13, 5, "Saturday", "May", 2], + ], + columns=[ + "date_id", + "year", + "month", + "day", + "day_of_week", + "day_name", + "month_name", + "quarter", + ], + ) with patch("src.dataframes.create_fact_payment") as mock_fp: with patch("src.dataframes.create_fact_purchase_orders") as mock_fpo: with patch("src.dataframes.create_fact_sales_order") as mock_fso: mock_fp.return_value = df_one mock_fpo.return_value = df_two mock_fso.return_value = df_three - result = create_dim_date({'dum':0}) - result.reset_index(inplace=True,drop=True) + result = create_dim_date({"dum": 0}) + result.reset_index(inplace=True, drop=True) assert result.eq(expected_df, axis="columns").all(axis=None) - -class TestCreateDimLocation: + +class TestCreateDimLocation: def test_returns_correct_columns_lo(self): - dict_df = {'address':pd.DataFrame(data=[['some_time','some_other_time',1,'SE18 9QO']], - columns=['created_at','last_updated','address_id','postal_code'])} + dict_df = { + "address": pd.DataFrame( + data=[["some_time", "some_other_time", 1, "SE18 9QO"]], + columns=["created_at", "last_updated", "address_id", "postal_code"], + ) + } result = create_dim_location(dict_df) - assert list(result.columns) == ['location_id','postal_code'] - + assert list(result.columns) == ["location_id", "postal_code"] + + class TestCreateDimTransaction: - def test_returns_correct_columns_tr(self): - dict_df = {'transaction':pd.DataFrame(data=[['some_time','some_other_time',1,'SE18 9QO']], - columns=['created_at','last_updated','transaction_id','some_other_id'])} + def test_returns_correct_columns_tr(self): + dict_df = { + "transaction": pd.DataFrame( + data=[["some_time", "some_other_time", 1, "SE18 9QO"]], + columns=[ + "created_at", + "last_updated", + "transaction_id", + "some_other_id", + ], + ) + } result = create_dim_transaction(dict_df) - assert list(result.columns) == ['transaction_id','some_other_id'] + assert list(result.columns) == ["transaction_id", "some_other_id"] + class TestCreateFactPayment: def test_returns_correct_columns_payment(self): - dict_df = {'payment':pd.DataFrame(data=[[dt.strptime('2022-11-03 14:20:49.962846','%Y-%m-%d %H:%M:%S.%f'), - dt.strptime('2022-12-14 16:20:49.962194','%Y-%m-%d %H:%M:%S.%f'),1,'SE18 9QO','2020-07-16']], - columns=['created_at','last_updated','payment_id','some_other_id','payment_date'])} - expected_cols = ['payment_record_id','created_date','created_time','last_updated_date', - 'last_updated_time','payment_date','payment_id','some_other_id'] + dict_df = { + "payment": pd.DataFrame( + data=[ + [ + dt.strptime( + "2022-11-03 14:20:49.962846", "%Y-%m-%d %H:%M:%S.%f" + ), + dt.strptime( + "2022-12-14 16:20:49.962194", "%Y-%m-%d %H:%M:%S.%f" + ), + 1, + "SE18 9QO", + "2020-07-16", + ] + ], + columns=[ + "created_at", + "last_updated", + "payment_id", + "some_other_id", + "payment_date", + ], + ) + } + expected_cols = [ + "payment_record_id", + "created_date", + "created_time", + "last_updated_date", + "last_updated_time", + "payment_date", + "payment_id", + "some_other_id", + ] result = create_fact_payment(dict_df) - assert isinstance(result,pd.DataFrame) + assert isinstance(result, pd.DataFrame) for col in list(result.columns): assert col in expected_cols for col in expected_cols: - if '_date' or '_time' in col: - assert result[col].dtype == 'O' - - \ No newline at end of file + if "_date" or "_time" in col: + assert result[col].dtype == "O" -- cgit v1.2.3 From 03787e3aabc5bc516bb7bfcc3831a74681932c36 Mon Sep 17 00:00:00 2001 From: HastarTara Date: Wed, 28 Aug 2024 09:48:07 +0100 Subject: moved extract_l & dataframes into own directory in src --- src/dataframes.py | 228 ------------------------------- src/transform_lambda.py | 217 ----------------------------- src/transform_lambda/dataframes.py | 228 +++++++++++++++++++++++++++++++ src/transform_lambda/transform_lambda.py | 217 +++++++++++++++++++++++++++++ 4 files changed, 445 insertions(+), 445 deletions(-) delete mode 100644 src/dataframes.py delete mode 100644 src/transform_lambda.py create mode 100644 src/transform_lambda/dataframes.py create mode 100644 src/transform_lambda/transform_lambda.py (limited to 'src/transform_lambda.py') diff --git a/src/dataframes.py b/src/dataframes.py deleted file mode 100644 index f122368..0000000 --- a/src/dataframes.py +++ /dev/null @@ -1,228 +0,0 @@ -import pandas as pd -from bs4 import BeautifulSoup -import requests - -# Table names: -# fact_sales_order -# fact_purchase_orders -# fact_payment -# dim_transaction -# dim_staff -# dim_payment_type -# dim_location -# dim_design -# dim_date -# dim_currency -# dim_counterparty - - -# no test, same as fact_payment -def create_fact_sales_order(dict_of_df): - df_sales = dict_of_df["sales_order"] - df_sales.index.name = "sales_record_id" - df_sales["created_date"] = 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 - - -# no test, same as fact_payment - - -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 - ) - df_po["last_updated_date"] = df_po["last_updated"].astype("datetime64[ns]").dt.date - df_po["last_updated_time"] = ( - df_po["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time - ) - df_po["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 = df_po.drop(labels=["created_at", "last_updated"], axis=1) - df_po.reset_index(inplace=True) - return df_po - - -# test passed - - -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 - ) - df_payment["created_time"] = ( - df_payment["created_at"].astype("datetime64[ns]").dt.floor("s").dt.time - ) - df_payment["last_updated_date"] = ( - df_payment["last_updated"].astype("datetime64[ns]").dt.date - ) - df_payment["last_updated_time"] = ( - df_payment["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time - ) - 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 - - -# 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 - - -# test passed - - -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"}) - ) - return df_loc - - -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="counterparty_legal_address_id", - how="outer", - ) - df_cp.drop( - columns=["legal_address_id", "counterparty_legal_address_id"], inplace=True - ) - return df_cp - - -# test passed - - -def create_dim_date(dict_of_df): - fact_dfs = [ - create_fact_payment(dict_of_df), - create_fact_purchase_orders(dict_of_df), - create_fact_sales_order(dict_of_df), - ] - list_of_date_columns = [] - for df in fact_dfs: - date_col_names = [ - col_name for col_name in list(df.columns) if "_date" in col_name - ] - 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"]) - 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 - return df_date - - -# tests passed - - -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"}) - ] - 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, names, left_on="currency_code", right_on="currency_code", how="inner" - ) - return dim_cur - - -# 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"]] - return dim_payment_type - - -# tests passed - - -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"] - ] - return dim_design - - -# tests passed - - -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", - ], - ] - return dim_staff diff --git a/src/transform_lambda.py b/src/transform_lambda.py deleted file mode 100644 index 93b2284..0000000 --- a/src/transform_lambda.py +++ /dev/null @@ -1,217 +0,0 @@ -import json -import boto3 -import re -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 - - -class DBConnectionException(Exception): - """Wraps pg8000.native Error or DatabaseError.""" - - def __init__(self, e): - """Initialise with provided error message.""" - self.message = str(e) - super().__init__(self.message) - - -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) - -TABLES = [ - "sales_order", - "transaction", - "payment", - "counterparty", - "address", - "staff", - "purchase_order", - "department", - "currency", - "design", - "payment_type", -] - - -def lambda_handler(event, context): - db = None - - try: - db = connect_to_database() - bucket = bucket_name("transform") - - existing_s3_files = list_existing_s3_files(bucket) - - dict_of_df = read_from_s3_subfolder_to_df( - TABLES, bucket_name("extract"), 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), - } - - 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), - } - - status = process_to_parquet_and_upload_to_s3( - existing_s3_files, immutable_df_dict, mutable_df_dict, bucket - ) - - if not status["uploaded"]: - logger.info("No dataframes written to the bucket.") - return { - "statusCode": 204, - "body": json.dumps("No files where uploaded."), - } - - return { - "statusCode": 200, - "body": json.dumps( - f"""Parquet files processed for {', '.join(status['uploaded'])} and uploaded successfully.{ - '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.")} - finally: - if db: - 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": []} - - for table_name, df in immutable_df_dict.items(): - 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 - # changed parquet_file variable to the file name - client.upload_file(f"{table_name}.parquet", 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(f"{table_name}.parquet", bucket, s3_key) - status["uploaded"].append(table_name) - - return status - - -def retrieve_secrets(): - secret_name = "bentley-secrets" - 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"] - - -def connect_to_database() -> Connection: - try: - secrets = json.loads(retrieve_secrets()) - host = secrets["host"] - port = secrets["port"] - user = secrets["user"] - password = secrets["password"] - database = secrets["database"] - - return Connection( - database=database, user=user, password=password, host=host, port=port - ) - except InterfaceError as i: - logger.error(f"Interface error: {i}") - raise DBConnectionException("Failed to connect to database") - - -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_df = [pd.read_csv(key) for key in list_of_keys] - 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"] - ] - - 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) - - if "Contents" in response: - existing_files = [obj["Key"] for obj in response["Contents"]] - else: - logger.error("The bucket is empty") - return [] # changed from None to [] so it is an iterable - - except ClientError as e: - logger.error(f"Error listing S3 objects: {e}") - raise e - - return existing_files - - -if __name__ == "__main__": - lambda_handler({}, "") diff --git a/src/transform_lambda/dataframes.py b/src/transform_lambda/dataframes.py new file mode 100644 index 0000000..f122368 --- /dev/null +++ b/src/transform_lambda/dataframes.py @@ -0,0 +1,228 @@ +import pandas as pd +from bs4 import BeautifulSoup +import requests + +# Table names: +# fact_sales_order +# fact_purchase_orders +# fact_payment +# dim_transaction +# dim_staff +# dim_payment_type +# dim_location +# dim_design +# dim_date +# dim_currency +# dim_counterparty + + +# no test, same as fact_payment +def create_fact_sales_order(dict_of_df): + df_sales = dict_of_df["sales_order"] + df_sales.index.name = "sales_record_id" + df_sales["created_date"] = 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 + + +# no test, same as fact_payment + + +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 + ) + df_po["last_updated_date"] = df_po["last_updated"].astype("datetime64[ns]").dt.date + df_po["last_updated_time"] = ( + df_po["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time + ) + df_po["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 = df_po.drop(labels=["created_at", "last_updated"], axis=1) + df_po.reset_index(inplace=True) + return df_po + + +# test passed + + +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 + ) + df_payment["created_time"] = ( + df_payment["created_at"].astype("datetime64[ns]").dt.floor("s").dt.time + ) + df_payment["last_updated_date"] = ( + df_payment["last_updated"].astype("datetime64[ns]").dt.date + ) + df_payment["last_updated_time"] = ( + df_payment["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time + ) + 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 + + +# 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 + + +# test passed + + +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"}) + ) + return df_loc + + +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="counterparty_legal_address_id", + how="outer", + ) + df_cp.drop( + columns=["legal_address_id", "counterparty_legal_address_id"], inplace=True + ) + return df_cp + + +# test passed + + +def create_dim_date(dict_of_df): + fact_dfs = [ + create_fact_payment(dict_of_df), + create_fact_purchase_orders(dict_of_df), + create_fact_sales_order(dict_of_df), + ] + list_of_date_columns = [] + for df in fact_dfs: + date_col_names = [ + col_name for col_name in list(df.columns) if "_date" in col_name + ] + 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"]) + 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 + return df_date + + +# tests passed + + +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"}) + ] + 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, names, left_on="currency_code", right_on="currency_code", how="inner" + ) + return dim_cur + + +# 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"]] + return dim_payment_type + + +# tests passed + + +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"] + ] + return dim_design + + +# tests passed + + +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", + ], + ] + return dim_staff diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py new file mode 100644 index 0000000..93b2284 --- /dev/null +++ b/src/transform_lambda/transform_lambda.py @@ -0,0 +1,217 @@ +import json +import boto3 +import re +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 + + +class DBConnectionException(Exception): + """Wraps pg8000.native Error or DatabaseError.""" + + def __init__(self, e): + """Initialise with provided error message.""" + self.message = str(e) + super().__init__(self.message) + + +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) + +TABLES = [ + "sales_order", + "transaction", + "payment", + "counterparty", + "address", + "staff", + "purchase_order", + "department", + "currency", + "design", + "payment_type", +] + + +def lambda_handler(event, context): + db = None + + try: + db = connect_to_database() + bucket = bucket_name("transform") + + existing_s3_files = list_existing_s3_files(bucket) + + dict_of_df = read_from_s3_subfolder_to_df( + TABLES, bucket_name("extract"), 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), + } + + 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), + } + + status = process_to_parquet_and_upload_to_s3( + existing_s3_files, immutable_df_dict, mutable_df_dict, bucket + ) + + if not status["uploaded"]: + logger.info("No dataframes written to the bucket.") + return { + "statusCode": 204, + "body": json.dumps("No files where uploaded."), + } + + return { + "statusCode": 200, + "body": json.dumps( + f"""Parquet files processed for {', '.join(status['uploaded'])} and uploaded successfully.{ + '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.")} + finally: + if db: + 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": []} + + for table_name, df in immutable_df_dict.items(): + 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 + # changed parquet_file variable to the file name + client.upload_file(f"{table_name}.parquet", 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(f"{table_name}.parquet", bucket, s3_key) + status["uploaded"].append(table_name) + + return status + + +def retrieve_secrets(): + secret_name = "bentley-secrets" + 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"] + + +def connect_to_database() -> Connection: + try: + secrets = json.loads(retrieve_secrets()) + host = secrets["host"] + port = secrets["port"] + user = secrets["user"] + password = secrets["password"] + database = secrets["database"] + + return Connection( + database=database, user=user, password=password, host=host, port=port + ) + except InterfaceError as i: + logger.error(f"Interface error: {i}") + raise DBConnectionException("Failed to connect to database") + + +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_df = [pd.read_csv(key) for key in list_of_keys] + 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"] + ] + + 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) + + if "Contents" in response: + existing_files = [obj["Key"] for obj in response["Contents"]] + else: + logger.error("The bucket is empty") + return [] # changed from None to [] so it is an iterable + + except ClientError as e: + logger.error(f"Error listing S3 objects: {e}") + raise e + + return existing_files + + +if __name__ == "__main__": + lambda_handler({}, "") -- cgit v1.2.3