From 8255a66c86dd932b8a453298067e0abb11fe639f Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Wed, 28 Aug 2024 11:14:09 +0100 Subject: test: remove aws_credentials parameter from fixtures --- tests/test_load_lambda.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 65106f7..da11c1e 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -20,13 +20,13 @@ def aws_credentials(): @pytest.fixture(scope="class") -def mock_s3_client(aws_credentials): +def mock_s3_client(): with mock_aws(): yield boto3.client("s3") @pytest.fixture(scope="class") -def mock_sm_client(aws_credentials): +def mock_sm_client(): with mock_aws(): yield boto3.client("secretsmanager") -- cgit v1.2.3 From 75d43f52a8ae7dd087f389de65b19501f2db4b3e Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Wed, 28 Aug 2024 11:14:57 +0100 Subject: test: remove unnecessary imports --- tests/test_load_lambda.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index da11c1e..830bb00 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -1,6 +1,4 @@ import pandas as pd -import pyarrow.parquet as pq -from io import BytesIO from moto import mock_aws import boto3 import botocore.exceptions -- cgit v1.2.3 From 23e20341e0b61c75ea81ab8a8d7a55bdd06e1b16 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Wed, 28 Aug 2024 11:16:09 +0100 Subject: test: import load_lambda functions individually --- tests/test_load_lambda.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 830bb00..3ad7158 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -4,7 +4,14 @@ import boto3 import botocore.exceptions import os import pytest -from src.load_lambda import * +from src.load_lambda import ( + lambda_handler, + retrieve_secrets, + connect_to_db_and_return_engine, + convert_parquet_files_to_dfs, + get_transform_bucket, + upload_dfs_to_database, +) import tempfile -- cgit v1.2.3 From d4c20088cdfc602faec21bd55df703a41dc0a343 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Wed, 28 Aug 2024 11:17:24 +0100 Subject: test: re-add json module import --- tests/test_load_lambda.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 3ad7158..02fd461 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -13,6 +13,7 @@ from src.load_lambda import ( upload_dfs_to_database, ) import tempfile +import json @pytest.fixture(scope="class") -- cgit v1.2.3 From b94cbcb355ad2bde408e5e714252d06621cbbc8e Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Wed, 28 Aug 2024 11:19:53 +0100 Subject: test: add staticmethod decorator to methods --- tests/test_load_lambda.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 02fd461..25a176d 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -38,7 +38,8 @@ def mock_sm_client(): class TestLambdaHandler: - def test_lambda_handler_returns_200_and_table_name_if_uploaded(self, mocker): + @staticmethod + def test_lambda_handler_returns_200_and_table_name_if_uploaded(mocker): mocker.patch( "src.load_lambda.upload_dfs_to_database", return_value={"uploaded": ["table_one", "table_two"], "not_uploaded": []}, @@ -48,7 +49,8 @@ class TestLambdaHandler: assert "table_one" in result["body"] assert "table_two" in result["body"] - def test_lambda_handler_returns_200_and_table_name_if_not_uploaded(self, mocker): + @staticmethod + def test_lambda_handler_returns_200_and_table_name_if_not_uploaded(mocker): mocker.patch( "src.load_lambda.upload_dfs_to_database", return_value={"uploaded": [], "not_uploaded": ["table_one"]}, @@ -57,7 +59,8 @@ class TestLambdaHandler: assert result["statusCode"] == 200 assert "No dataframes were uploaded" in result["body"] - def test_lambda_handler_returns_error_if_both_lists_empty(self, mocker): + @staticmethod + def test_lambda_handler_returns_error_if_both_lists_empty(mocker): mocker.patch( "src.load_lambda.upload_dfs_to_database", return_value={"uploaded": [], "not_uploaded": []}, -- cgit v1.2.3 From 5ae0d01c0429833e61e36c73d2fad4aef565ddd2 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Wed, 28 Aug 2024 12:02:21 +0100 Subject: wip: add tests for upload_dfs_to_database --- tests/test_load_lambda.py | 35 +++++++++++++++++++++++++++++++---- 1 file changed, 31 insertions(+), 4 deletions(-) diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 25a176d..cb27178 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -199,7 +199,34 @@ class TestConvertParquetToDfs: class TestUploadDfsToDatabase: - # Full success test - # Partial success test - # Failure test - pass + def test_function_returns_dictionary_with_uploaded_and_not_uploaded_keys( + self, mocker + ): + mocker.patch( + "src.load_lambda.convert_parquet_files_to_dfs", + return_value={"dim_counterparty.parquet": pd.DataFrame()}, + ) + mocker.patch( + "src.load_lambda.connect_to_db_and_return_engine", + return_value="test_engine", + ) + + result = upload_dfs_to_database() + + assert "uploaded" in result + assert "not_uploaded" in result + + def test_function_returns_uploaded_and_not_uploaded_tables(self, mocker): + mocker.patch( + "src.load_lambda.convert_parquet_files_to_dfs", + return_value={"dim_counterparty.parquet": pd.DataFrame()}, + ) + mocker.patch( + "src.load_lambda.connect_to_db_and_return_engine", + return_value="test_engine", + ) + + result = upload_dfs_to_database() + + assert result["uploaded"] == ["dim_counterparty"] + assert result["not_uploaded"] == [] -- cgit v1.2.3 From f109d5b2856efe6b2822e4376fd06b773fa5c4a1 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Wed, 28 Aug 2024 12:02:54 +0100 Subject: chore: remove commented code --- tests/test_load_lambda.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index cb27178..290576f 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -160,11 +160,6 @@ class TestConvertParquetToDfs: ) assert result == {} - # def test_function_returns_dictionary_with_table_with_file_key(): - # # need to mock parquet file and upload to mock bucket - # result = convert_parquet_files_to_dfs(bucket_name="transform_bucket", client=mock_s3_client) - # assert "dim_staff" in result - def test_function_returns_dictionary_with_file_key_and_dataframe( self, mock_s3_client ): -- cgit v1.2.3 From c05fd0ed48b84b35b030c96d1e067496ebe71e84 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Wed, 28 Aug 2024 12:05:11 +0100 Subject: test: add staticmethod decorator to remaining methods --- tests/test_load_lambda.py | 40 ++++++++++++++++++++++++---------------- 1 file changed, 24 insertions(+), 16 deletions(-) diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 290576f..cc6a85e 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -72,7 +72,8 @@ class TestLambdaHandler: class TestRetrieveSecrets: - def test_retrieve_secrets_returns_dictionary(self, mock_sm_client): + @staticmethod + def test_retrieve_secrets_returns_dictionary(mock_sm_client): secret = { "cohort_id": "test_cohort_id", "user": "test_user_id", @@ -90,7 +91,8 @@ class TestRetrieveSecrets: assert isinstance(result, dict) - def test_retrieve_secrets_returns_correct_keys_and_values(self, mock_sm_client): + @staticmethod + def test_retrieve_secrets_returns_correct_keys_and_values(mock_sm_client): secret_name = "test_secret" result = json.loads(retrieve_secrets(mock_sm_client, secret_name)) @@ -98,7 +100,8 @@ class TestRetrieveSecrets: assert result["user"] == "test_user_id" assert result["password"] == "test_password" - def test_retrieve_secrets_returns_client_error_if_no_secret(self, mock_sm_client): + @staticmethod + def test_retrieve_secrets_returns_client_error_if_no_secret(mock_sm_client): secret_name = "another_test_secret" with pytest.raises(botocore.exceptions.ClientError) as error: @@ -106,7 +109,8 @@ class TestRetrieveSecrets: class TestConnectToDBAndReturnEngine: - def test_returns_unsuccessful_connection_when_wrong_credentials(self): + @staticmethod + def test_returns_unsuccessful_connection_when_wrong_credentials(): sm_secret = { "host": "host", "port": "port", @@ -120,11 +124,13 @@ class TestConnectToDBAndReturnEngine: class TestGetTransformBucket: - def test_raises_value_error_if_no_buckets(self, mock_s3_client): + @staticmethod + def test_raises_value_error_if_no_buckets(mock_s3_client): with pytest.raises(ValueError, match="No transform bucket found"): get_transform_bucket(mock_s3_client) - def test_raises_value_error_if_no_transform_bucket(self, mock_s3_client): + @staticmethod + def test_raises_value_error_if_no_transform_bucket(mock_s3_client): mock_s3_client.create_bucket( Bucket="extract_bucket", CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, @@ -132,7 +138,8 @@ class TestGetTransformBucket: with pytest.raises(ValueError, match="No transform bucket found"): get_transform_bucket(mock_s3_client) - def test_returns_transform_bucket_if_one_bucket(self, mock_s3_client): + @staticmethod + def test_returns_transform_bucket_if_one_bucket(mock_s3_client): mock_s3_client.create_bucket( Bucket="transform_bucket", CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, @@ -140,7 +147,8 @@ class TestGetTransformBucket: result = get_transform_bucket(mock_s3_client) assert result == "transform_bucket" - def test_only_returns_transform_bucket_if_several_buckets(self, mock_s3_client): + @staticmethod + def test_only_returns_transform_bucket_if_several_buckets(mock_s3_client): mock_s3_client.create_bucket( Bucket="another_test_bucket", CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, @@ -150,7 +158,8 @@ class TestGetTransformBucket: class TestConvertParquetToDfs: - def test_function_returns_empty_dictionary_if_no_files(self, mock_s3_client): + @staticmethod + def test_function_returns_empty_dictionary_if_no_files(mock_s3_client): mock_s3_client.create_bucket( Bucket="transform_bucket", CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, @@ -160,9 +169,8 @@ class TestConvertParquetToDfs: ) assert result == {} - def test_function_returns_dictionary_with_file_key_and_dataframe( - self, mock_s3_client - ): + @staticmethod + def test_function_returns_dictionary_with_file_key_and_dataframe(mock_s3_client): with tempfile.TemporaryDirectory() as tmp: d = { "test": ["Hello", "Bye"], @@ -194,9 +202,8 @@ class TestConvertParquetToDfs: class TestUploadDfsToDatabase: - def test_function_returns_dictionary_with_uploaded_and_not_uploaded_keys( - self, mocker - ): + @staticmethod + def test_function_returns_dictionary_with_uploaded_and_not_uploaded_keys(mocker): mocker.patch( "src.load_lambda.convert_parquet_files_to_dfs", return_value={"dim_counterparty.parquet": pd.DataFrame()}, @@ -211,7 +218,8 @@ class TestUploadDfsToDatabase: assert "uploaded" in result assert "not_uploaded" in result - def test_function_returns_uploaded_and_not_uploaded_tables(self, mocker): + @staticmethod + def test_function_returns_uploaded_and_not_uploaded_tables(mocker): mocker.patch( "src.load_lambda.convert_parquet_files_to_dfs", return_value={"dim_counterparty.parquet": pd.DataFrame()}, -- cgit v1.2.3 From c89de263ec3f79c68d9b7a35d2bffcd4fb62a8c4 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Wed, 28 Aug 2024 12:34:02 +0100 Subject: test: update upload_dfs_to_database tests --- tests/test_load_lambda.py | 52 ++++++++++++++++++++++------------------------- 1 file changed, 24 insertions(+), 28 deletions(-) diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index cc6a85e..b284588 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -14,6 +14,7 @@ from src.load_lambda import ( ) import tempfile import json +from unittest.mock import MagicMock, patch @pytest.fixture(scope="class") @@ -202,34 +203,29 @@ class TestConvertParquetToDfs: class TestUploadDfsToDatabase: - @staticmethod - def test_function_returns_dictionary_with_uploaded_and_not_uploaded_keys(mocker): - mocker.patch( - "src.load_lambda.convert_parquet_files_to_dfs", - return_value={"dim_counterparty.parquet": pd.DataFrame()}, - ) - mocker.patch( - "src.load_lambda.connect_to_db_and_return_engine", - return_value="test_engine", - ) - - result = upload_dfs_to_database() - - assert "uploaded" in result - assert "not_uploaded" in result + @pytest.fixture + def mock_engine(self): + engine = MagicMock() + engine.dispose = MagicMock() + return engine + + @pytest.fixture + def mock_df(self): + df = MagicMock(spec=pd.DataFrame) + df.to_sql = MagicMock() + return df @staticmethod - def test_function_returns_uploaded_and_not_uploaded_tables(mocker): - mocker.patch( + def test_function_returns_dictionary_with_uploaded_and_not_uploaded_keys( + mock_engine, mock_df + ): + with patch( "src.load_lambda.convert_parquet_files_to_dfs", - return_value={"dim_counterparty.parquet": pd.DataFrame()}, - ) - mocker.patch( - "src.load_lambda.connect_to_db_and_return_engine", - return_value="test_engine", - ) - - result = upload_dfs_to_database() - - assert result["uploaded"] == ["dim_counterparty"] - assert result["not_uploaded"] == [] + return_value={"dim_counterparty.parquet": mock_df}, + ), patch( + "src.load_lambda.connect_to_db_and_return_engine", return_value=mock_engine + ): + result = upload_dfs_to_database() + + assert "uploaded" in result + assert "not_uploaded" in result -- cgit v1.2.3 From 03a5959df25f74d52ed5393c2a5af6b1b9eb34c9 Mon Sep 17 00:00:00 2001 From: T-Aji Date: Wed, 28 Aug 2024 12:48:13 +0100 Subject: refactored functs to include columns instead of drop columns --- src/load_lambda.py | 5 +- src/transform_lambda/dataframes.py | 157 ++++++++++++++++++++----------- src/transform_lambda/transform_lambda.py | 5 +- tests/test_dataframes.py | 2 +- 4 files changed, 111 insertions(+), 58 deletions(-) diff --git a/src/load_lambda.py b/src/load_lambda.py index 7339ab9..926b4db 100644 --- a/src/load_lambda.py +++ b/src/load_lambda.py @@ -134,6 +134,9 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): file_obj = client.get_object(Bucket=bucket_name, Key=file_key) parquet_file = pq.ParquetFile(BytesIO(file_obj["Body"].read())) df = parquet_file.read().to_pandas() + print("df", df) + print("type", type(df)) + print(df.columns) dfs[file_key] = df except ClientError as e: logger.error(f"Unable to retrieve S3 object {file_key}: {e}") @@ -148,7 +151,7 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): except ClientError as client_error: logger.error(f"Unable to list objects: {client_error}") raise - + print() return dfs diff --git a/src/transform_lambda/dataframes.py b/src/transform_lambda/dataframes.py index 2a46bd6..bf0556b 100644 --- a/src/transform_lambda/dataframes.py +++ b/src/transform_lambda/dataframes.py @@ -37,30 +37,28 @@ def create_fact_sales_order(dict_of_df): df_sales["agreed_payment_date"] = pd.to_datetime( df_sales["agreed_payment_date"], format="%Y-%m-%d" ) - df_sales = df_sales.drop(labels=["created_at", "last_updated"], axis=1) - - df_sales.reset_index(inplace=True) - return df_sales + fact_sales = df_sales.loc[:, + [ + "sales_order_id", + "created_date", + "created_time", + "last_updated_date", + "last_updated_time", + "staff_id", + "counterparty_id", + "units_sold", + "unit_price", + "currency_id", + "design_id", + "agreed_payment_date", + "agreed_delivery_date", + "agreed_delivery_location_id" + ], + ] + fact_sales.rename(columns={"staff_id": "sales_staff_id"}).reset_index(inplace=True) + - df_sales["created_date"] = df_sales["created_at"].astype("datetime64[ns]").dt.date - df_sales["created_time"] = ( - df_sales["created_at"].astype("datetime64[ns]").dt.floor("s").dt.time - ) - df_sales["last_updated_date"] = ( - df_sales["last_updated"].astype("datetime64[ns]").dt.date - ) - df_sales["last_updated_time"] = ( - df_sales["last_updated"].astype("datetime64[ns]").dt.floor("s").dt.time - ) - df_sales["agreed_delivery_date"] = pd.to_datetime( - df_sales["agreed_delivery_date"], format="%Y-%m-%d" - ) - df_sales["agreed_payment_date"] = pd.to_datetime( - df_sales["agreed_payment_date"], format="%Y-%m-%d" - ) - df_sales = df_sales.drop(labels=["created_at", "last_updated"], axis=1) - df_sales.reset_index(inplace=True) - return df_sales + return fact_sales # no test, same as fact_payment @@ -83,9 +81,27 @@ def create_fact_purchase_orders(dict_of_df): df_po["agreed_payment_date"] = pd.to_datetime( df_po["agreed_payment_date"], format="%Y-%m-%d" ) - df_po = df_po.drop(labels=["created_at", "last_updated"], axis=1) - df_po.reset_index(inplace=True) - return df_po + fact_purchase_order = df_po.loc[:, + [ + "purchase_order_id", + "created_date", + "created_time", + "last_updated_date", + "last_updated_time", + "staff_id", + "counterparty_id", + "item_code", + "item_quantity", + "item_unit_price", + "currency_id", + "agreed_delivery_date", + "agreed_payment_date", + "agreed_delivery_location_id" + ] + + ] + fact_purchase_order.reset_index(inplace=True) + return fact_purchase_order # test passed @@ -109,38 +125,57 @@ def create_fact_payment(dict_of_df): df_payment["payment_date"] = pd.to_datetime( df_payment["payment_date"], format="%Y-%m-%d" ) - df_payment = df_payment.drop(labels=["created_at", "last_updated"], axis=1) - - df_payment.reset_index(inplace=True) - return df_payment + fact_payment = df_payment.loc[:, + [ + "payment_id", + "created_date", + "created_time", + "last_updated_date", + "last_updated_time", + "transaction_id", + "counterparty_id", + "payment_amount", + "currency_id", + "payment_type_id", + "paid", + "payment_date" + ] + ] + fact_payment.reset_index(inplace=True) + return fact_payment # test passed def create_dim_transaction(dict_of_df): - df_transaction = dict_of_df["transaction"].drop( - labels=["created_at", "last_updated"], axis=1 - ) - return df_transaction + dim_transaction = dict_of_df["transaction"].loc[:, + [ + "transaction_id", + "transaction_type", + "sales_order_id", + "purchase_order_id" + ] + ] + return dim_transaction # test passed def create_dim_location(dict_of_df): - df_loc = ( - dict_of_df["address"] - .drop(labels=["created_at", "last_updated"], axis=1) + dim_location = ( + dict_of_df["address"].drop(labels=["created_at", "last_updated"], axis=1) .rename(columns={"address_id": "location_id"}) ) - return df_loc + return dim_location def create_dim_counterparty(dict_of_df): df_prefixed_address = ( dict_of_df["address"] .drop(labels=["created_at", "last_updated"], axis=1) + .rename(columns={"phone": "phone_number"}) .add_prefix("counterparty_legal_", axis=1) ) df_cp = pd.merge( @@ -149,15 +184,18 @@ def create_dim_counterparty(dict_of_df): left_on="legal_address_id", right_on="counterparty_legal_address_id", how="inner", - ) - df_cp.drop( - columns=[ + )#.dropna(inplace=True) + dim_counterparty = df_cp.drop( + labels=[ "legal_address_id", "counterparty_legal_address_id", - ], - inplace=True, + "created_at", + "last_updated", + "commercial_contact", + "delivery_contact" + ], axis=1 ) - return df_cp + return dim_counterparty # test passed @@ -179,6 +217,7 @@ def create_dim_date(dict_of_df): sr_date = pd.array(pd.concat(list_of_date_columns), dtype="datetime64[ns]") df_date = pd.DataFrame(data=sr_date, columns=["date_id"]) df_date.drop_duplicates(inplace=True) + # df_date.dropna(inplace=True) df_date["year"] = df_date["date_id"].dt.year df_date["month"] = df_date["date_id"].dt.month df_date["day"] = df_date["date_id"].dt.day @@ -210,10 +249,11 @@ def scrape_currency_names(): def create_dim_currency(dict_of_df, names=scrape_currency_names()): df_cur = dict_of_df["currency"].drop(labels=["created_at", "last_updated"], axis=1) - dim_cur = pd.merge( - df_cur, names, left_on="currency_code", right_on="currency_code", how="inner" + dim_currency = pd.merge( + df_cur, names, left_on="currency_code", right_on="currency_code", how="left" ) - return dim_cur + dim_currency.drop_duplicates(inplace=True) + return dim_currency # tests passed @@ -221,7 +261,12 @@ def create_dim_currency(dict_of_df, names=scrape_currency_names()): def create_dim_payment_type(dict_of_df): df_payment_type = dict_of_df["payment_type"] - dim_payment_type = df_payment_type.loc[:, ["payment_type_id", "payment_type_name"]] + dim_payment_type = df_payment_type.loc[:, + [ + "payment_type_id", + "payment_type_name" + ] + ] return dim_payment_type @@ -230,8 +275,13 @@ def create_dim_payment_type(dict_of_df): def create_dim_design(dict_of_df): df_design = dict_of_df["design"] - dim_design = df_design.loc[ - :, ["design_id", "design_name", "file_name", "file_location"] + dim_design = df_design.loc[:, + [ + "design_id", + "design_name", + "file_name", + "file_location" + ] ] return dim_design @@ -243,15 +293,14 @@ def create_dim_staff(dict_of_df): staff_department = pd.merge( dict_of_df["staff"], dict_of_df["department"], on="department_id", how="left" ) - dim_staff = staff_department.loc[ - :, + dim_staff = staff_department.loc[:, [ "staff_id", "first_name", "last_name", "department_name", "location", - "email_address", - ], + "email_address" + ] ] return dim_staff diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py index 93b2284..1453c6c 100644 --- a/src/transform_lambda/transform_lambda.py +++ b/src/transform_lambda/transform_lambda.py @@ -42,7 +42,7 @@ TABLES = [ "department", "currency", "design", - "payment_type", + "payment_type" ] @@ -73,7 +73,8 @@ def lambda_handler(event, context): "fact_payment": create_fact_payment(dict_of_df), "dim_currency": create_dim_currency(dict_of_df), } - + print(immutable_df_dict.values()) + print(mutable_df_dict.values()) status = process_to_parquet_and_upload_to_s3( existing_s3_files, immutable_df_dict, mutable_df_dict, bucket ) diff --git a/tests/test_dataframes.py b/tests/test_dataframes.py index ea7bad1..7dd592a 100644 --- a/tests/test_dataframes.py +++ b/tests/test_dataframes.py @@ -1,4 +1,4 @@ -from src.dataframes import * +from src.transform_lambda.dataframes import * import pandas as pd from unittest.mock import patch from datetime import datetime as dt -- cgit v1.2.3 From d064b2ec2c7393f8de50560a7edfe100851bfea3 Mon Sep 17 00:00:00 2001 From: T-Aji Date: Wed, 28 Aug 2024 14:39:13 +0100 Subject: debugging load_lambda --- src/load_lambda.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/load_lambda.py b/src/load_lambda.py index 926b4db..272cb8c 100644 --- a/src/load_lambda.py +++ b/src/load_lambda.py @@ -164,13 +164,13 @@ def upload_dfs_to_database(): "dim_date.parquet", # this needs to be mutable "dim_location.parquet", "dim_staff.parquet", - "dim_design.parquet", + "dim_design.parquet" ] mutable_df_dict = [ "fact_sales_order", "fact_purchase_order", "fact_payment", - "dim_currency", + "dim_currency" ] for file_name, df in dict_of_dfs.items(): @@ -182,6 +182,7 @@ def upload_dfs_to_database(): df.to_sql( table_name, con=db_engine, + schema="project_team_2", if_exists="append", index=False, ) -- cgit v1.2.3 From 6235a2bb04b60d57a41196b07bbf0296920c6980 Mon Sep 17 00:00:00 2001 From: T-Aji 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(-) 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, "<<<<<< Date: Wed, 28 Aug 2024 22:46:00 +0100 Subject: fix: adds missing dataframes and resolves tables upload to end data warehouse in case the table is empty --- .gitignore | 6 +++++- src/load_lambda.py | 24 +++++++++++++----------- src/transform_lambda/dataframes.py | 19 ++++++++++++++----- src/transform_lambda/transform_lambda.py | 4 +++- 4 files changed, 35 insertions(+), 18 deletions(-) diff --git a/.gitignore b/.gitignore index 6aa03fc..480ae4b 100644 --- a/.gitignore +++ b/.gitignore @@ -14,4 +14,8 @@ __pycache__/ # OS-Related Files .DS_Store -venv \ No newline at end of file +venv + +#files +/dim_* +/fact_* \ No newline at end of file diff --git a/src/load_lambda.py b/src/load_lambda.py index cdcf105..8f921b8 100644 --- a/src/load_lambda.py +++ b/src/load_lambda.py @@ -161,18 +161,15 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): else: continue immutables_l = list(set(immutables_l)) - print(mutables_d,'mutables_d') latest_s3_keys = [] for k,v in mutables_d.items(): latest_s3_keys.append(dt.strftime(get_latest_timestamp(v), f"{k}/%Y/%m/%d/{k}_%H:%M:%S.parquet")) - print(latest_s3_keys,'latest') - print(immutables_l,'immutables_l') - for file_key in latest_s3_keys+immutables_l: + for file_key in immutables_l+latest_s3_keys: try: file_obj = client.get_object(Bucket=bucket_name, Key=file_key) parquet_file = pq.ParquetFile(BytesIO(file_obj["Body"].read())) df = parquet_file.read().to_pandas() - df_without_nulls = df.dropna() + df_without_nulls = df.dropna(how='all') #>> can't do 'any' (default) because we lose rows in dim_location #print("df_without_nulls", df_without_nulls) #print("type", type(df_without_nulls)) #print(df_without_nulls.columns) @@ -202,12 +199,14 @@ def upload_dfs_to_database(): # "dim_date.parquet", # this needs to be mutable # "dim_location.parquet", # "dim_staff.parquet", - # "dim_design.parquet" + # "dim_design.parquet", + # 'dim_transaction.parquet' #This one was missing, + 'dim_payment_type.parquet' ] mutable_df_dict = [ - "dim_currency", - "fact_sales_order", - "fact_purchase_order", + # "dim_currency", + # "fact_sales_order", + # "fact_purchase_order", "fact_payment" ] @@ -215,7 +214,9 @@ def upload_dfs_to_database(): for file_name, df in dict_of_dfs.items(): print(df.dtypes, "dtypes") print(df.head()) - if file_name in immutable_df_dict: + print(file_name,"<<< FILE NAME") + print(immutable_df_dict,"<< Date: Thu, 29 Aug 2024 09:47:58 +0100 Subject: fix: added comma. Code complete and uploads all tables in one go if no data exists per each table --- src/load_lambda.py | 19 +++++++++---------- 1 file changed, 9 insertions(+), 10 deletions(-) diff --git a/src/load_lambda.py b/src/load_lambda.py index 8f921b8..941ae97 100644 --- a/src/load_lambda.py +++ b/src/load_lambda.py @@ -57,7 +57,6 @@ def retrieve_secrets(client=None, secret_name=None): try: get_secret_value_response = client.get_secret_value(SecretId=secret_name) - print(get_secret_value_response) except ClientError as e: logger.error(f"Failed to retrieve secret {secret_name}: {str(e)}", exc_info=True) raise e @@ -195,18 +194,18 @@ def upload_dfs_to_database(): dict_of_dfs = convert_parquet_files_to_dfs() db_engine = connect_to_db_and_return_engine() immutable_df_dict = [ - # #"dim_counterparty.parquet", - # "dim_date.parquet", # this needs to be mutable - # "dim_location.parquet", - # "dim_staff.parquet", - # "dim_design.parquet", - # 'dim_transaction.parquet' #This one was missing, + "dim_counterparty.parquet", + "dim_date.parquet", # this needs to be mutable + "dim_location.parquet", + "dim_staff.parquet", + "dim_design.parquet", + 'dim_transaction.parquet', #This one was missing, 'dim_payment_type.parquet' ] mutable_df_dict = [ - # "dim_currency", - # "fact_sales_order", - # "fact_purchase_order", + "dim_currency", + "fact_sales_order", + "fact_purchase_order", "fact_payment" ] -- cgit v1.2.3 From 42ad135b25044bb1c7ab8a553f038c8da9de0f75 Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Thu, 29 Aug 2024 08:57:48 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 48e7dae according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/107 --- src/load_lambda.py | 78 +++++++++++++-------- src/transform_lambda/dataframes.py | 116 ++++++++++++++----------------- src/transform_lambda/transform_lambda.py | 6 +- 3 files changed, 105 insertions(+), 95 deletions(-) diff --git a/src/load_lambda.py b/src/load_lambda.py index 941ae97..86189dc 100644 --- a/src/load_lambda.py +++ b/src/load_lambda.py @@ -58,10 +58,14 @@ def retrieve_secrets(client=None, secret_name=None): try: get_secret_value_response = client.get_secret_value(SecretId=secret_name) except ClientError as e: - logger.error(f"Failed to retrieve secret {secret_name}: {str(e)}", exc_info=True) + logger.error( + f"Failed to retrieve secret {secret_name}: {str(e)}", exc_info=True + ) raise e except KeyError: - logger.error(f"Secret {secret_name} does not contain a SecretString", exc_info=True) + logger.error( + f"Secret {secret_name} does not contain a SecretString", exc_info=True + ) raise ValueError(f"Secret {secret_name} does not contain a SecretString") return get_secret_value_response["SecretString"] @@ -117,6 +121,7 @@ def get_transform_bucket(client=None): # convert parquet files into dataframes # return a dictionary of dataframes with name as key, and dataframe object as value + def get_latest_timestamp(existing_files): if existing_files: all_datetimes = [] @@ -124,19 +129,17 @@ def get_latest_timestamp(existing_files): match = re.search(r"\/(.+/).+_(.+)\.parquet", file_name) if match: datetime_str = "".join(match.group(1, 2)) - all_datetimes.append( - dt.strptime(datetime_str, "%Y/%m/%d/%H:%M:%S") - ) + all_datetimes.append(dt.strptime(datetime_str, "%Y/%m/%d/%H:%M:%S")) return max(all_datetimes) if all_datetimes else dt.min return existing_files + def convert_parquet_files_to_dfs(bucket_name=None, client=None): mutable_df_dict = [ "dim_currency", "fact_sales_order", "fact_purchase_order", - "fact_payment" - + "fact_payment", ] try: @@ -145,12 +148,12 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): if bucket_name is None: bucket_name = get_transform_bucket() files = client.list_objects_v2(Bucket=bucket_name) - + dfs = {} if "Contents" in files: - s3_key_list = [file["Key"]for file in files["Contents"]] + s3_key_list = [file["Key"] for file in files["Contents"]] immutables_l = [] - mutables_d = {prefix:[] for prefix in mutable_df_dict} + mutables_d = {prefix: [] for prefix in mutable_df_dict} for tab, s3_key in mutables_d.items(): for file in s3_key_list: if tab in file: @@ -161,22 +164,31 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): continue immutables_l = list(set(immutables_l)) latest_s3_keys = [] - for k,v in mutables_d.items(): - latest_s3_keys.append(dt.strftime(get_latest_timestamp(v), f"{k}/%Y/%m/%d/{k}_%H:%M:%S.parquet")) - for file_key in immutables_l+latest_s3_keys: + for k, v in mutables_d.items(): + latest_s3_keys.append( + dt.strftime( + get_latest_timestamp(v), f"{k}/%Y/%m/%d/{k}_%H:%M:%S.parquet" + ) + ) + for file_key in immutables_l + latest_s3_keys: try: file_obj = client.get_object(Bucket=bucket_name, Key=file_key) parquet_file = pq.ParquetFile(BytesIO(file_obj["Body"].read())) df = parquet_file.read().to_pandas() - df_without_nulls = df.dropna(how='all') #>> can't do 'any' (default) because we lose rows in dim_location - #print("df_without_nulls", df_without_nulls) - #print("type", type(df_without_nulls)) - #print(df_without_nulls.columns) + # >> can't do 'any' (default) because we lose rows in dim_location + df_without_nulls = df.dropna(how="all") + # print("df_without_nulls", df_without_nulls) + # print("type", type(df_without_nulls)) + # print(df_without_nulls.columns) dfs[file_key] = df_without_nulls except ClientError as e: - logger.error(f"Unable to retrieve S3 object {file_key}: {e}", exc_info=True) + logger.error( + f"Unable to retrieve S3 object {file_key}: {e}", exc_info=True + ) except Exception as e: - logger.error(f"Unable to process file {file_key}: {e}", exc_info=True) + logger.error( + f"Unable to process file {file_key}: {e}", exc_info=True + ) else: logger.error(f"No files found in {bucket_name}.", exc_info=True) return {} @@ -199,23 +211,22 @@ def upload_dfs_to_database(): "dim_location.parquet", "dim_staff.parquet", "dim_design.parquet", - 'dim_transaction.parquet', #This one was missing, - 'dim_payment_type.parquet' + "dim_transaction.parquet", # This one was missing, + "dim_payment_type.parquet", ] mutable_df_dict = [ "dim_currency", "fact_sales_order", "fact_purchase_order", - "fact_payment" - + "fact_payment", ] with db_engine.begin() as connection: for file_name, df in dict_of_dfs.items(): print(df.dtypes, "dtypes") print(df.head()) - print(file_name,"<<< FILE NAME") - print(immutable_df_dict,"<< Date: Thu, 29 Aug 2024 10:21:31 +0100 Subject: ci: create deploy.yml --- .github/workflows/deploy.yml | 42 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) create mode 100644 .github/workflows/deploy.yml diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml new file mode 100644 index 0000000..5672048 --- /dev/null +++ b/.github/workflows/deploy.yml @@ -0,0 +1,42 @@ +name: deploy-terraform + +on: + pull_request: + branches: + - main + push: + branches: + - main + + +jobs: + deploy-terraform: + name: Deploy Terraform + runs-on: ubuntu-latest + #needs: run-checks (must ref on-commit.yml file) + environment: production + steps: + - name: Checkout Repo + uses: actions/checkout@v4 + + - name: Install Terraform + uses: hashicorp/setup-terraform@v3 + + - name: Configure AWS Credentials + uses: aws-actions/configure-aws-credentials@v4 + with: + aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} + aws-region: ${{ secrets.AWS_REGION }} + + - name: Terraform Init + working-directory: terraform + run: terraform init + + - name: Terraform Plan + working-directory: terraform + run: terraform plan + + - name: Terraform Apply + working-directory: terraform + run: terraform apply --auto-approve -- cgit v1.2.3 From 67bef521e6ff8012b425c55bb18cf29d8a41fef9 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 10:35:16 +0100 Subject: infra(tf): fix layer paths --- terraform/lambda.tf | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/terraform/lambda.tf b/terraform/lambda.tf index 5f4a58e..b6f36fb 100644 --- a/terraform/lambda.tf +++ b/terraform/lambda.tf @@ -86,8 +86,8 @@ resource "aws_lambda_function" "extract_lambda" { data "archive_file" "transform_lambda_zip" { type = "zip" - source_dir = "${path.module}../src/transform_lambda" - output_path = "${path.module}../transform_lambda.zip" + source_dir = "${path.module}/../src/transform_lambda" + output_path = "${path.module}/../transform_lambda.zip" } resource "aws_s3_object" "transform_lambda_code" { -- cgit v1.2.3 From 54aca43830f7c4e6962eabc2aace63256e81eb96 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 11:12:43 +0100 Subject: infra(tf): update load function_name & lambda_function_arn --- terraform/events.tf | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/terraform/events.tf b/terraform/events.tf index 53ae10a..7f8f641 100644 --- a/terraform/events.tf +++ b/terraform/events.tf @@ -86,7 +86,7 @@ resource "aws_s3_bucket_notification" "extract_bucket_notification" { resource "aws_lambda_permission" "allow_s3_transform_bucket" { statement_id = "AllowS3InvokeLambdaTransform${random_string.s3_transform_suffix.result}" action = "lambda:InvokeFunction" - function_name = aws_lambda_function.transform_lambda.function_name + function_name = aws_lambda_function.load_lambda.function_name principal = "s3.amazonaws.com" source_arn = aws_s3_bucket.transform_bucket.arn @@ -102,7 +102,7 @@ resource "aws_s3_bucket_notification" "transform_bucket_notification" { lambda_function { events = ["s3:ObjectCreated:*"] - lambda_function_arn = aws_lambda_function.transform_lambda.arn + lambda_function_arn = aws_lambda_function.load_lambda.arn } depends_on = [aws_lambda_permission.allow_s3_transform_bucket] -- cgit v1.2.3 From cd7a8f83dc20c045f607aba9ea0e83e0ef22b19e Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 11:16:57 +0100 Subject: chore: remove car_data.parquet --- car_data.parquet | Bin 2827 -> 0 bytes 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 car_data.parquet diff --git a/car_data.parquet b/car_data.parquet deleted file mode 100644 index 1853af6..0000000 Binary files a/car_data.parquet and /dev/null differ -- cgit v1.2.3 From e3f6e9b7477ac633d272d2fcacb13e42bb7eb572 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 12:19:21 +0100 Subject: feat: separate layer zip generation for lambda functions --- scripts/make_layer_zip_01.sh | 8 ++++++++ scripts/make_layer_zip_02.sh | 9 +++++++++ 2 files changed, 17 insertions(+) create mode 100755 scripts/make_layer_zip_01.sh create mode 100755 scripts/make_layer_zip_02.sh diff --git a/scripts/make_layer_zip_01.sh b/scripts/make_layer_zip_01.sh new file mode 100755 index 0000000..145ea82 --- /dev/null +++ b/scripts/make_layer_zip_01.sh @@ -0,0 +1,8 @@ +# Description: Make the zip file for the layer for the extract lambda function + +cd "$(dirname "$0")/.." +mkdir -p python/lib/python3.11/site-packages +pip3 install --upgrade -r requirements.txt -t python/lib/python3.11/site-packages +rm layer.zip +zip -r layer.zip python +rm -r python/ diff --git a/scripts/make_layer_zip_02.sh b/scripts/make_layer_zip_02.sh new file mode 100755 index 0000000..53e9099 --- /dev/null +++ b/scripts/make_layer_zip_02.sh @@ -0,0 +1,9 @@ +# Description: Make the zip file for the layer for the transform & load +# lambda functions + +cd "$(dirname "$0")/.." +mkdir -p python/lib/python3.11/site-packages +pip3 install --upgrade -r requirements.txt -t python/lib/python3.11/site-packages +rm layer.zip +zip -r layer.zip python +rm -r python/ -- cgit v1.2.3 From 4838efc2cdff9966d4aa2042068be744e4ae1f84 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 12:19:45 +0100 Subject: chore: rm old make_layer_zip.sh script --- scripts/make_layer_zip.sh | 8 -------- 1 file changed, 8 deletions(-) delete mode 100755 scripts/make_layer_zip.sh diff --git a/scripts/make_layer_zip.sh b/scripts/make_layer_zip.sh deleted file mode 100755 index eabe301..0000000 --- a/scripts/make_layer_zip.sh +++ /dev/null @@ -1,8 +0,0 @@ -# Description: Make the zip file for the layer - -cd "$(dirname "$0")/.." -mkdir -p python/lib/python3.11/site-packages -pip3 install --upgrade -r requirements.txt -t python/lib/python3.11/site-packages -rm layer.zip -zip -r layer.zip python -rm -r python/ -- cgit v1.2.3 From 107619511b8d1668707048d9caf38d4f6da96426 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 12:22:24 +0100 Subject: refactor: separate layer zip prep --- scripts/make_layer_zip_01.sh | 10 +++++----- scripts/make_layer_zip_02.sh | 10 +++++----- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/scripts/make_layer_zip_01.sh b/scripts/make_layer_zip_01.sh index 145ea82..c8b4d2a 100755 --- a/scripts/make_layer_zip_01.sh +++ b/scripts/make_layer_zip_01.sh @@ -1,8 +1,8 @@ # Description: Make the zip file for the layer for the extract lambda function cd "$(dirname "$0")/.." -mkdir -p python/lib/python3.11/site-packages -pip3 install --upgrade -r requirements.txt -t python/lib/python3.11/site-packages -rm layer.zip -zip -r layer.zip python -rm -r python/ +mkdir -p python_01/lib/python3.11/site-packages +pip3 install --upgrade -r requirements_01.txt -t python_01/lib/python3.11/site-packages +rm layer_01.zip +zip -r layer_01.zip python_01 +rm -r python_01/ diff --git a/scripts/make_layer_zip_02.sh b/scripts/make_layer_zip_02.sh index 53e9099..8b5bb33 100755 --- a/scripts/make_layer_zip_02.sh +++ b/scripts/make_layer_zip_02.sh @@ -2,8 +2,8 @@ # lambda functions cd "$(dirname "$0")/.." -mkdir -p python/lib/python3.11/site-packages -pip3 install --upgrade -r requirements.txt -t python/lib/python3.11/site-packages -rm layer.zip -zip -r layer.zip python -rm -r python/ +mkdir -p python_02/lib/python3.11/site-packages +pip3 install --upgrade -r requirements_02.txt -t python_02/lib/python3.11/site-packages +rm layer_02.zip +zip -r layer_02.zip python_02 +rm -r python_02/ -- cgit v1.2.3 From ecf330af27b4b32535de61fd3dd958bd5be6d278 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 12:33:05 +0100 Subject: chore: add new requirements.txt files for extract/transform/lambda layers --- requirements_lambda_01.txt | 3 +++ requirements_lambda_02.txt | 9 +++++++++ 2 files changed, 12 insertions(+) create mode 100644 requirements_lambda_01.txt create mode 100644 requirements_lambda_02.txt diff --git a/requirements_lambda_01.txt b/requirements_lambda_01.txt new file mode 100644 index 0000000..10f56be --- /dev/null +++ b/requirements_lambda_01.txt @@ -0,0 +1,3 @@ +boto3 +botocore +pg8000 \ No newline at end of file diff --git a/requirements_lambda_02.txt b/requirements_lambda_02.txt new file mode 100644 index 0000000..20c88d7 --- /dev/null +++ b/requirements_lambda_02.txt @@ -0,0 +1,9 @@ +pandas +pyarrow +SQLAlchemy +auto_mix_prep +beautifulsoup4 +boto3 +botocore +pg8000 +Requests \ No newline at end of file -- cgit v1.2.3 From abcf3df1913507517f8e2401eb9a79033b755287 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 12:34:21 +0100 Subject: chore: update requirements file names in bash scripts --- scripts/make_layer_zip_01.sh | 2 +- scripts/make_layer_zip_02.sh | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/make_layer_zip_01.sh b/scripts/make_layer_zip_01.sh index c8b4d2a..c2d4534 100755 --- a/scripts/make_layer_zip_01.sh +++ b/scripts/make_layer_zip_01.sh @@ -2,7 +2,7 @@ cd "$(dirname "$0")/.." mkdir -p python_01/lib/python3.11/site-packages -pip3 install --upgrade -r requirements_01.txt -t python_01/lib/python3.11/site-packages +pip3 install --upgrade -r requirements_lambda_01.txt -t python_01/lib/python3.11/site-packages rm layer_01.zip zip -r layer_01.zip python_01 rm -r python_01/ diff --git a/scripts/make_layer_zip_02.sh b/scripts/make_layer_zip_02.sh index 8b5bb33..c788acf 100755 --- a/scripts/make_layer_zip_02.sh +++ b/scripts/make_layer_zip_02.sh @@ -3,7 +3,7 @@ cd "$(dirname "$0")/.." mkdir -p python_02/lib/python3.11/site-packages -pip3 install --upgrade -r requirements_02.txt -t python_02/lib/python3.11/site-packages +pip3 install --upgrade -r requirements_lambda_02.txt -t python_02/lib/python3.11/site-packages rm layer_02.zip zip -r layer_02.zip python_02 rm -r python_02/ -- cgit v1.2.3 From 62417b57133cc1c6df3efd89c247706a93e3f182 Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 17:46:33 +0100 Subject: infra(tf): restore make_layer_zip & add layer_01/layer_02 creation in same script --- scripts/make_layer_zip.sh | 17 +++++++++++++++++ scripts/make_layer_zip_01.sh | 8 -------- scripts/make_layer_zip_02.sh | 9 --------- 3 files changed, 17 insertions(+), 17 deletions(-) create mode 100755 scripts/make_layer_zip.sh delete mode 100755 scripts/make_layer_zip_01.sh delete mode 100755 scripts/make_layer_zip_02.sh diff --git a/scripts/make_layer_zip.sh b/scripts/make_layer_zip.sh new file mode 100755 index 0000000..7f64873 --- /dev/null +++ b/scripts/make_layer_zip.sh @@ -0,0 +1,17 @@ +# Description: Make the zip file for the layer for the extract lambda function + +cd "$(dirname "$0")/.." + +# Layer 01 +mkdir -p python/lib/python3.11/site-packages +pip3 install --upgrade -r requirements_lambda_01.txt -t python/lib/python3.11/site-packages +rm layer_01.zip +zip -r layer_01.zip python +rm -r python/ + +# Layer 02 +mkdir -p python/lib/python3.11/site-packages +pip3 install --upgrade -r requirements_lambda_02.txt -t python/lib/python3.11/site-packages +rm layer_02.zip +zip -r layer_02.zip python +rm -r python/ diff --git a/scripts/make_layer_zip_01.sh b/scripts/make_layer_zip_01.sh deleted file mode 100755 index c2d4534..0000000 --- a/scripts/make_layer_zip_01.sh +++ /dev/null @@ -1,8 +0,0 @@ -# Description: Make the zip file for the layer for the extract lambda function - -cd "$(dirname "$0")/.." -mkdir -p python_01/lib/python3.11/site-packages -pip3 install --upgrade -r requirements_lambda_01.txt -t python_01/lib/python3.11/site-packages -rm layer_01.zip -zip -r layer_01.zip python_01 -rm -r python_01/ diff --git a/scripts/make_layer_zip_02.sh b/scripts/make_layer_zip_02.sh deleted file mode 100755 index c788acf..0000000 --- a/scripts/make_layer_zip_02.sh +++ /dev/null @@ -1,9 +0,0 @@ -# Description: Make the zip file for the layer for the transform & load -# lambda functions - -cd "$(dirname "$0")/.." -mkdir -p python_02/lib/python3.11/site-packages -pip3 install --upgrade -r requirements_lambda_02.txt -t python_02/lib/python3.11/site-packages -rm layer_02.zip -zip -r layer_02.zip python_02 -rm -r python_02/ -- cgit v1.2.3 From 904c75c88b9eb2a5f28b5fc737d253f70196a17b Mon Sep 17 00:00:00 2001 From: Alex Schofield Date: Thu, 29 Aug 2024 17:47:25 +0100 Subject: infra(tf)/wip: update lambda layer creation scripting --- terraform/lambda.tf | 81 +++++++++++++++++++++++++++++++++-------------------- 1 file changed, 50 insertions(+), 31 deletions(-) diff --git a/terraform/lambda.tf b/terraform/lambda.tf index b6f36fb..1e12180 100644 --- a/terraform/lambda.tf +++ b/terraform/lambda.tf @@ -3,46 +3,65 @@ #################### locals { - layer_dir = "../" - layer_zip = "layer.zip" - layer_name = "lambda_layer" - script_dir = "../scripts" - layer_zip_path = "${local.layer_dir}/${local.layer_zip}" + layer_dir = "../" + layer_zip_01 = "layer_01.zip" + layer_zip_02 = "layer_02.zip" + layer_name_01 = "lambda_layer_01" + layer_name_02 = "lambda_layer_02" + script_dir = "../scripts" + layer_zip_01_path = "${local.layer_dir}${local.layer_zip_01}" + layer_zip_02_path = "${local.layer_dir}${local.layer_zip_02}" } -###################### -# Lambda Layer Setup # -###################### - resource "null_resource" "prepare_layer" { - - # New change: only run the script if the layer zip does not exist - + provisioner "local-exec" { + command = "bash ${local.script_dir}/make_layer_zip.sh" + } triggers = { - layer_zip_exists = fileexists(local.layer_zip_path) ? "exists" : "not_exists" + always_run = timestamp() } +} - provisioner "local-exec" { - command = "if [ ! -f ${local.layer_zip_path} ]; then bash ${local.script_dir}/make_layer_zip.sh; fi" - } +################################ +# Lambda Layer (Extract) Setup # +################################ +resource "aws_s3_object" "lambda_layer_zip_01" { + bucket = aws_s3_bucket.lambda_code_bucket.id #bucket instead of id + key = "${local.layer_name_01}/${local.layer_zip_01}" + source = "${local.layer_dir}${local.layer_zip_01}" + depends_on = [null_resource.prepare_layer] + etag = fileexists(local.layer_zip_01_path) ? filemd5(local.layer_zip_01_path) : null + force_destroy = true +} + +resource "aws_lambda_layer_version" "lambda_layer_01" { + layer_name = local.layer_name_01 + compatible_runtimes = ["python3.11"] + s3_bucket = aws_s3_bucket.lambda_code_bucket.bucket + s3_key = aws_s3_object.lambda_layer_zip_01.key + source_code_hash = fileexists(local.layer_zip_01_path) ? filebase64sha256(local.layer_zip_01_path) : null + depends_on = [aws_s3_object.lambda_layer_zip_01] } -resource "aws_s3_object" "lambda_layer_zip" { - bucket = aws_s3_bucket.lambda_code_bucket.id #bucket instead of id - key = "${local.layer_name}/${local.layer_zip}" - source = "${local.layer_dir}/${local.layer_zip}" - depends_on = [null_resource.prepare_layer] - etag = fileexists(local.layer_zip_path) ? filemd5(local.layer_zip_path) : null +######################################### +# Lambda Layer (Load & Transform) Setup # +######################################### +resource "aws_s3_object" "lambda_layer_zip_02" { + bucket = aws_s3_bucket.lambda_code_bucket.id #bucket instead of id + key = "${local.layer_name_02}/${local.layer_zip_02}" + source = "${local.layer_dir}${local.layer_zip_02}" + depends_on = [null_resource.prepare_layer] + etag = fileexists(local.layer_zip_02_path) ? filemd5(local.layer_zip_02_path) : null + force_destroy = true } -resource "aws_lambda_layer_version" "lambda_layer" { - layer_name = local.layer_name +resource "aws_lambda_layer_version" "lambda_layer_02" { + layer_name = local.layer_name_02 compatible_runtimes = ["python3.11"] s3_bucket = aws_s3_bucket.lambda_code_bucket.bucket - s3_key = aws_s3_object.lambda_layer_zip.key - source_code_hash = fileexists(local.layer_zip_path) ? filebase64sha256(local.layer_zip_path) : null - skip_destroy = true - depends_on = [aws_s3_object.lambda_layer_zip] + s3_key = aws_s3_object.lambda_layer_zip_02.key + source_code_hash = fileexists(local.layer_zip_02_path) ? filebase64sha256(local.layer_zip_02_path) : null + depends_on = [aws_s3_object.lambda_layer_zip_02] } ########################### @@ -65,7 +84,7 @@ resource "aws_lambda_function" "extract_lambda" { function_name = var.extract_lambda_name s3_bucket = aws_s3_bucket.lambda_code_bucket.bucket s3_key = aws_s3_object.extract_lambda_code.key - layers = [aws_lambda_layer_version.lambda_layer.arn] + layers = [aws_lambda_layer_version.lambda_layer_01.arn] role = aws_iam_role.multi_service_role.arn handler = "extract_lambda.lambda_handler" runtime = "python3.11" @@ -101,7 +120,7 @@ resource "aws_lambda_function" "transform_lambda" { function_name = var.transform_lambda_name s3_bucket = aws_s3_bucket.lambda_code_bucket.bucket s3_key = aws_s3_object.transform_lambda_code.key - layers = [aws_lambda_layer_version.lambda_layer.arn] + layers = [aws_lambda_layer_version.lambda_layer_02.arn] role = aws_iam_role.multi_service_role.arn handler = "transform_lambda.lambda_handler" runtime = "python3.11" @@ -135,7 +154,7 @@ resource "aws_lambda_function" "load_lambda" { function_name = var.load_lambda_name s3_bucket = aws_s3_bucket.lambda_code_bucket.bucket s3_key = aws_s3_object.load_lambda_code.key - layers = [aws_lambda_layer_version.lambda_layer.arn] + layers = [aws_lambda_layer_version.lambda_layer_02.arn] role = aws_iam_role.multi_service_role.arn handler = "load_lambda.lambda_handler" runtime = "python3.11" -- cgit v1.2.3 From 8be9b9b5c06a166064b1606b9e72390953cf10fa Mon Sep 17 00:00:00 2001 From: Alex Date: Fri, 30 Aug 2024 13:28:51 +0100 Subject: Update README.md --- README.md | 20 +++++++------------- 1 file changed, 7 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 7d7e499..e8e1ebd 100644 --- a/README.md +++ b/README.md @@ -36,16 +36,10 @@ can: We aim for the project to have certain features. Some are more prioritised than others. -- [ ] Automated data ingestion from `totesys` db -- [ ] Data storage for ingested and processed data in S3 buckets -- [ ] Data transformation for data warehouse schema -- [ ] Automated data loading into the data warehouse schema -- [ ] Logging and monitoring with CloudWatch -- [ ] Notifications for errors and successful runs (e.g. successful ingestion) -- [ ] Visualisation of warehouse data - -# Test Coverage -TBA - -# Contributors -TBA +- Automated data ingestion from `totesys` db +- Data storage for ingested and processed data in S3 buckets +- Data transformation for data warehouse schema +- Automated data loading into the data warehouse schema +- Logging and monitoring with CloudWatch +- Notifications for errors and successful runs (e.g. successful ingestion) +- Visualisation of warehouse data -- cgit v1.2.3