From 828e8292440d4395fbb00afff4e35ff194f07a95 Mon Sep 17 00:00:00 2001 From: Ellie Date: Thu, 22 Aug 2024 16:56:15 +0100 Subject: wip: add test file for load lambda --- tests/test_load_lambda.py | 9 +++++++++ 1 file changed, 9 insertions(+) create mode 100644 tests/test_load_lambda.py (limited to 'tests/test_load_lambda.py') diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py new file mode 100644 index 0000000..0572340 --- /dev/null +++ b/tests/test_load_lambda.py @@ -0,0 +1,9 @@ +import boto3 +import pandas as pd +import pyarrow.parquet as pq +from io import BytesIO +from src.load_lambda import convert_parquet_files_to_dataframes + +class TestConvertParquetToDFs: + def test_convert_parquet_to_dfs_returns_df(): + \ No newline at end of file -- cgit v1.2.3 From 09c8191ce983e4335cfb131d21ddb5413b849cfb Mon Sep 17 00:00:00 2001 From: Ellie Date: Fri, 23 Aug 2024 11:18:24 +0100 Subject: add tests --- src/load_lambda.py | 61 ++++++++++++++++++++++++++++++++++++++++++++--- tests/test_load_lambda.py | 3 +-- 2 files changed, 59 insertions(+), 5 deletions(-) (limited to 'tests/test_load_lambda.py') diff --git a/src/load_lambda.py b/src/load_lambda.py index a3fd996..d95c27a 100644 --- a/src/load_lambda.py +++ b/src/load_lambda.py @@ -4,6 +4,9 @@ import pandas as pd import pyarrow.parquet as pq from io import BytesIO import logging +import json +from src.extract_lambda import retrieve_secrets, connect_to_database +from sqlalchemy import create_engine logger = logging.getLogger(__name__) @@ -17,6 +20,43 @@ logging.basicConfig( logging.getLogger("botocore").setLevel(logging.WARNING) +def lambda_handler(event, context): + db = None + try: + uploaded_tables = upload_dfs_to_database() + if uploaded_tables == []: + return { + "statusCode": 200, + "body": json.dumps("No datframes were uploaded."), + } + return { + "statusCode": 200, + "body": json.dumps( + f"""The following dataframes were uploaded successfully: + {', '.join(upload_dfs_to_database['updated'])}.""" + ), + } + except Exception as e: + logger.error(f"Error: {e}", exc_info=True) + return {"statusCode": 500, "body": json.dumps("Internal server error.")} + finally: + if db: + db.close() + +# connect to database, slightly different way of doing it, to allow manipulation through pandas +def connect_to_db_and_return_engine(): + secrets = json.loads(retrieve_secrets("bentley-RDS-credentials")) #need to amend retrieve secrets function + host = secrets["host"] + port = secrets["port"] + user = secrets["user"] + password = secrets["password"] + database = secrets["database"] + conn_str = f'postgresql+pg8000://{user}:{password}@{host}:{port}/{database}' + engine = create_engine(conn_str) #interface between python (pandas) and SQL + return engine + + + # get transform bucket def transform_bucket(client=None): if client is None: @@ -41,7 +81,7 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): bucket_name = transform_bucket(client) files = client.list_objects_v2(Bucket=bucket_name) - dfs = [] + dfs = {} if "Contents" in files: for file in files["Contents"]: file_key = file['Key'] @@ -49,7 +89,7 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): file_obj = client.get_object(Bucket=bucket_name, Key=file_key) parquet_file = pq.ParquetFile(BytesIO(file_obj['Body'].read())) df = parquet_file.read().to_pandas() - dfs.append(df) + dfs[file_key] = df except ClientError as e: logger.error(f"Unable to retrieve S3 object {file_key}: {e}") except Exception as e: @@ -64,4 +104,19 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): logger.error(f"Unable to list objects: {client_error}") raise - return dfs + return dfs + +def upload_dfs_to_database(): + uploaded = [] + dict_of_dfs = convert_parquet_files_to_dfs() + db_engine = connect_to_db_and_return_engine() + try: + for table_name, df in dict_of_dfs: + df.to_sql(table_name, con=db_engine, ifexists="replace", index=False) + uploaded.append(table_name) + except Exception as e: + logger.error(f"Error uploading dataframes: {e}") + db_engine.dispose() + return uploaded + + # aiming to return a list of uploaded tables \ No newline at end of file diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 0572340..d9ea918 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -1,8 +1,7 @@ -import boto3 import pandas as pd import pyarrow.parquet as pq from io import BytesIO -from src.load_lambda import convert_parquet_files_to_dataframes +from src.load_lambda import convert_parquet_files_to_dfs class TestConvertParquetToDFs: def test_convert_parquet_to_dfs_returns_df(): -- cgit v1.2.3 From f3bb705a31ab9d94dc856c2de0da4b7b73a57fae Mon Sep 17 00:00:00 2001 From: Ellie Date: Fri, 23 Aug 2024 12:38:25 +0100 Subject: add get transform bucket test --- src/load_lambda.py | 2 +- tests/test_load_lambda.py | 48 +++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 45 insertions(+), 5 deletions(-) (limited to 'tests/test_load_lambda.py') diff --git a/src/load_lambda.py b/src/load_lambda.py index f92bb45..a9d5ac5 100644 --- a/src/load_lambda.py +++ b/src/load_lambda.py @@ -1,5 +1,5 @@ import boto3 -from botocore.exceptions import ClientError, InterfaceError +from botocore.exceptions import ClientError import pandas as pd import pyarrow.parquet as pq from io import BytesIO diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index d9ea918..2392f10 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -1,8 +1,48 @@ import pandas as pd import pyarrow.parquet as pq from io import BytesIO -from src.load_lambda import convert_parquet_files_to_dfs +from moto import mock_aws +import boto3 +import os +import pytest +from src.load_lambda import lambda_handler, connect_to_db_and_return_engine, get_transform_bucket, convert_parquet_files_to_dfs, upload_dfs_to_database -class TestConvertParquetToDFs: - def test_convert_parquet_to_dfs_returns_df(): - \ No newline at end of file +@pytest.fixture(scope="class") +def aws_credentials(): + os.environ["AWS_ACCESS_KEY_ID"] = "testing" + os.environ["AWS_SECRET_ACCESS_KEY"] = "testing" + os.environ["AWS_SECURIT_TOKEN"] = "testing" + os.environ["AWS_SESSION_TOKEN"] = "testing" + os.environ["AWS_DEFAULT_REGION"] = "eu-west-2" + + +@pytest.fixture(scope="class") +def s3_client(aws_credentials): + with mock_aws(): + yield boto3.client("s3") + +@pytest.fixture(scope="function") +def s3_mock_bucket(s3_client): + bucket = s3_client.create_bucket( + Bucket="transform_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + return bucket + + +class TestLambdaHandler: + pass + +class TestConnectToDBAndReturnEngine: + pass + +class TestGetTransformBucket: + def test_get_transform_bucket_returns_string(self, s3_client, s3_mock_bucket): + result = get_transform_bucket(s3_client) + assert result == "transform_bucket" + +class TestConvertParquetToDfs: + pass + +class TestUploadDfsToDatabase: + pass \ No newline at end of file -- cgit v1.2.3 From 2e85e8f14f35bebb7e96a9dff7bc59ebaefe32f6 Mon Sep 17 00:00:00 2001 From: Ellie Date: Fri, 23 Aug 2024 13:15:35 +0100 Subject: adds passing transform bucket tests --- tests/test_load_lambda.py | 30 +++++++++++++++++++----------- 1 file changed, 19 insertions(+), 11 deletions(-) (limited to 'tests/test_load_lambda.py') diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 2392f10..7f001df 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -17,18 +17,10 @@ def aws_credentials(): @pytest.fixture(scope="class") -def s3_client(aws_credentials): +def mock_s3_client(aws_credentials): with mock_aws(): yield boto3.client("s3") -@pytest.fixture(scope="function") -def s3_mock_bucket(s3_client): - bucket = s3_client.create_bucket( - Bucket="transform_bucket", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) - return bucket - class TestLambdaHandler: pass @@ -37,8 +29,24 @@ class TestConnectToDBAndReturnEngine: pass class TestGetTransformBucket: - def test_get_transform_bucket_returns_string(self, s3_client, s3_mock_bucket): - result = get_transform_bucket(s3_client) + def test_get_transform_bucket_raises_error_if_no_buckets(self, mock_s3_client): + with pytest.raises(ValueError, match="No transform bucket found"): + get_transform_bucket(mock_s3_client) + + def test_get_transform_bucket_returns_transform_bucket_if_one_bucket(self, mock_s3_client): + mock_s3_client.create_bucket( + Bucket="transform_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + result = get_transform_bucket(mock_s3_client) + assert result == "transform_bucket" + + def test_get_transform_bucket_only_returns_transform_bucket_if_several_buckets(self, mock_s3_client): + mock_s3_client.create_bucket( + Bucket="extract_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + result = get_transform_bucket(mock_s3_client) assert result == "transform_bucket" class TestConvertParquetToDfs: -- cgit v1.2.3 From 0c95b93303dea04e18aefe57e3b6fef7e4127c3c Mon Sep 17 00:00:00 2001 From: Ellie Date: Fri, 23 Aug 2024 13:22:23 +0100 Subject: add working completed tests for get transform bucket --- tests/test_load_lambda.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) (limited to 'tests/test_load_lambda.py') diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 7f001df..f1c2b01 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -29,11 +29,19 @@ class TestConnectToDBAndReturnEngine: pass class TestGetTransformBucket: - def test_get_transform_bucket_raises_error_if_no_buckets(self, mock_s3_client): + def test_raises_value_error_if_no_buckets(self, mock_s3_client): with pytest.raises(ValueError, match="No transform bucket found"): get_transform_bucket(mock_s3_client) - def test_get_transform_bucket_returns_transform_bucket_if_one_bucket(self, mock_s3_client): + def test_raises_value_error_if_no_transform_bucket(self, mock_s3_client): + mock_s3_client.create_bucket( + Bucket="extract_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + with pytest.raises(ValueError, match="No transform bucket found"): + get_transform_bucket(mock_s3_client) + + def test_returns_transform_bucket_if_one_bucket(self, mock_s3_client): mock_s3_client.create_bucket( Bucket="transform_bucket", CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, @@ -41,16 +49,16 @@ class TestGetTransformBucket: result = get_transform_bucket(mock_s3_client) assert result == "transform_bucket" - def test_get_transform_bucket_only_returns_transform_bucket_if_several_buckets(self, mock_s3_client): + def test_only_returns_transform_bucket_if_several_buckets(self, mock_s3_client): mock_s3_client.create_bucket( - Bucket="extract_bucket", + Bucket="another_test_bucket", CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, ) result = get_transform_bucket(mock_s3_client) assert result == "transform_bucket" class TestConvertParquetToDfs: - pass + pass class TestUploadDfsToDatabase: pass \ No newline at end of file -- cgit v1.2.3 From e26b7be8331d89826fbf95e1b1bd4fe88186c307 Mon Sep 17 00:00:00 2001 From: Ellie Date: Fri, 23 Aug 2024 17:04:29 +0100 Subject: add updated tests --- tests/test_load_lambda.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) (limited to 'tests/test_load_lambda.py') diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index f1c2b01..3e42c2a 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -25,6 +25,9 @@ def mock_s3_client(aws_credentials): class TestLambdaHandler: pass +class TestRetrieveSecrets: + pass + class TestConnectToDBAndReturnEngine: pass @@ -58,7 +61,18 @@ class TestGetTransformBucket: assert result == "transform_bucket" class TestConvertParquetToDfs: - pass + def test_function_returns_empty_dictionary_if_no_files(self, mock_s3_client): + mock_s3_client.create_bucket( + Bucket="transform_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + result = convert_parquet_files_to_dfs(bucket_name="transform_bucket", client=mock_s3_client) + 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 class TestUploadDfsToDatabase: pass \ No newline at end of file -- cgit v1.2.3 From 0ff29566a1eb9551bb83bcc07705c932d22f8c08 Mon Sep 17 00:00:00 2001 From: Ellie Date: Fri, 23 Aug 2024 17:06:59 +0100 Subject: add updated test --- tests/test_load_lambda.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'tests/test_load_lambda.py') diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index 3e42c2a..e04ccec 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -69,10 +69,10 @@ class TestConvertParquetToDfs: result = convert_parquet_files_to_dfs(bucket_name="transform_bucket", client=mock_s3_client) assert result == {} - def test_function_returns_dictionary_with_table_with_file_key(): - # 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_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 class TestUploadDfsToDatabase: pass \ No newline at end of file -- cgit v1.2.3 From 69edb14dad584d45fa6a83a90c08292b84795507 Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Fri, 23 Aug 2024 16:11:45 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 0ff2956 according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/95 --- src/load_lambda.py | 75 ++++++++++++++++++++++++++++++++--------------- tests/test_load_lambda.py | 44 +++++++++++++++++---------- 2 files changed, 80 insertions(+), 39 deletions(-) (limited to 'tests/test_load_lambda.py') diff --git a/src/load_lambda.py b/src/load_lambda.py index 8eaea32..6e6bc80 100644 --- a/src/load_lambda.py +++ b/src/load_lambda.py @@ -40,6 +40,7 @@ def lambda_handler(event, context): logger.error(f"Error: {e}", exc_info=True) return {"statusCode": 500, "body": json.dumps("Internal server error.")} + def retrieve_secrets(): secret_name = "bentley-RDS-credentials" region_name = "eu-west-2" @@ -59,7 +60,10 @@ def retrieve_secrets(): return get_secret_value_response["SecretString"] + # connect to database, slightly different way of doing it, to allow manipulation through pandas + + def connect_to_db_and_return_engine(): try: secrets = json.loads(retrieve_secrets()) @@ -68,13 +72,14 @@ def connect_to_db_and_return_engine(): user = secrets["user"] password = secrets["password"] database = secrets["database"] - conn_str = f'postgresql+pg8000://{user}:{password}@{host}:{port}/{database}' - engine = create_engine(conn_str) #interface between python (pandas) and SQL + conn_str = f"postgresql+pg8000://{user}:{password}@{host}:{port}/{database}" + # interface between python (pandas) and SQL + engine = create_engine(conn_str) return engine except Exception as e: logger.error(f"Interface error: {e}") raise RuntimeError("Failed to create database engine") - + # get transform bucket def get_transform_bucket(client=None): @@ -85,9 +90,11 @@ def get_transform_bucket(client=None): except ClientError as e: logger.error(f"Error listing S3 buckets: {e}") raise RuntimeError("Error listing S3 buckets") - + transform_bucket_filter = [ - bucket["Name"] for bucket in response["Buckets"] if "transform" in bucket["Name"] + bucket["Name"] + for bucket in response["Buckets"] + if "transform" in bucket["Name"] ] if not transform_bucket_filter: @@ -96,9 +103,12 @@ def get_transform_bucket(client=None): return transform_bucket_filter[0] + # list and then retrieve parquet files from S3 bucket # convert parquet files into dataframes -# return a dictionary of dataframes with name as key, and dataframe object as value +# return a dictionary of dataframes with name as key, and dataframe object as value + + def convert_parquet_files_to_dfs(bucket_name=None, client=None): try: if client is None: @@ -110,10 +120,10 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): dfs = {} if "Contents" in files: for file in files["Contents"]: - file_key = file['Key'] + file_key = file["Key"] try: file_obj = client.get_object(Bucket=bucket_name, Key=file_key) - parquet_file = pq.ParquetFile(BytesIO(file_obj['Body'].read())) + parquet_file = pq.ParquetFile(BytesIO(file_obj["Body"].read())) df = parquet_file.read().to_pandas() dfs[file_key] = df except ClientError as e: @@ -132,34 +142,51 @@ def convert_parquet_files_to_dfs(bucket_name=None, client=None): return dfs + def upload_dfs_to_database(): upload_status = {"uploaded": [], "not_uploaded": []} dict_of_dfs = convert_parquet_files_to_dfs() db_engine = connect_to_db_and_return_engine() - immutable_df_dict = ["dim_counterparty.parquet", - "dim_date.parquet", #this needs to be mutable - "dim_location.parquet", - "dim_staff.parquet", - "dim_design.parquet"] - mutable_df_dict = ["fact_sales_order", - "fact_purchase_order", - "fact_payment", - "dim_currency"] - + immutable_df_dict = [ + "dim_counterparty.parquet", + "dim_date.parquet", # this needs to be mutable + "dim_location.parquet", + "dim_staff.parquet", + "dim_design.parquet", + ] + mutable_df_dict = [ + "fact_sales_order", + "fact_purchase_order", + "fact_payment", + "dim_currency", + ] + for file_name, df in dict_of_dfs.items(): if file_name in immutable_df_dict: table_name = file_name.split(".")[0] try: - df.to_sql(table_name, con=db_engine, schema="project_team_2", if_exists="overwrite", index=False) + df.to_sql( + table_name, + con=db_engine, + schema="project_team_2", + if_exists="overwrite", + index=False, + ) upload_status["uploaded"].append(table_name) except Exception as e: logger.error(f"Error uploading dataframe {file_name} to database: {e}") raise - elif file_name.rsplit('_', 1)[0] in mutable_df_dict: - table_name = file_name.rsplit('_', 1)[0] + elif file_name.rsplit("_", 1)[0] in mutable_df_dict: + table_name = file_name.rsplit("_", 1)[0] try: - df.to_sql(table_name, con=db_engine, schema="project_team_2", if_exists="overwrite", index=False) - upload_status["uploaded"].append(table_name) + df.to_sql( + table_name, + con=db_engine, + schema="project_team_2", + if_exists="overwrite", + index=False, + ) + upload_status["uploaded"].append(table_name) except Exception as e: logger.error(f"Error uploading dataframe {file_name} to database: {e}") raise @@ -167,4 +194,4 @@ def upload_dfs_to_database(): upload_status["not_uploaded"].append(file_name) logger.error(f"{file_name} does not correspond with table in database") db_engine.dispose() - return upload_status \ No newline at end of file + return upload_status diff --git a/tests/test_load_lambda.py b/tests/test_load_lambda.py index e04ccec..88c71e4 100644 --- a/tests/test_load_lambda.py +++ b/tests/test_load_lambda.py @@ -5,7 +5,14 @@ from moto import mock_aws import boto3 import os import pytest -from src.load_lambda import lambda_handler, connect_to_db_and_return_engine, get_transform_bucket, convert_parquet_files_to_dfs, upload_dfs_to_database +from src.load_lambda import ( + lambda_handler, + connect_to_db_and_return_engine, + get_transform_bucket, + convert_parquet_files_to_dfs, + upload_dfs_to_database, +) + @pytest.fixture(scope="class") def aws_credentials(): @@ -25,12 +32,15 @@ def mock_s3_client(aws_credentials): class TestLambdaHandler: pass + class TestRetrieveSecrets: pass + class TestConnectToDBAndReturnEngine: pass + class TestGetTransformBucket: def test_raises_value_error_if_no_buckets(self, mock_s3_client): with pytest.raises(ValueError, match="No transform bucket found"): @@ -38,35 +48,38 @@ class TestGetTransformBucket: def test_raises_value_error_if_no_transform_bucket(self, mock_s3_client): mock_s3_client.create_bucket( - Bucket="extract_bucket", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) + Bucket="extract_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) with pytest.raises(ValueError, match="No transform bucket found"): get_transform_bucket(mock_s3_client) def test_returns_transform_bucket_if_one_bucket(self, mock_s3_client): mock_s3_client.create_bucket( - Bucket="transform_bucket", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) + Bucket="transform_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) result = get_transform_bucket(mock_s3_client) assert result == "transform_bucket" def test_only_returns_transform_bucket_if_several_buckets(self, mock_s3_client): mock_s3_client.create_bucket( - Bucket="another_test_bucket", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) + Bucket="another_test_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) result = get_transform_bucket(mock_s3_client) assert result == "transform_bucket" + class TestConvertParquetToDfs: def test_function_returns_empty_dictionary_if_no_files(self, mock_s3_client): mock_s3_client.create_bucket( - Bucket="transform_bucket", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) - result = convert_parquet_files_to_dfs(bucket_name="transform_bucket", client=mock_s3_client) + Bucket="transform_bucket", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + result = convert_parquet_files_to_dfs( + bucket_name="transform_bucket", client=mock_s3_client + ) assert result == {} # def test_function_returns_dictionary_with_table_with_file_key(): @@ -74,5 +87,6 @@ class TestConvertParquetToDfs: # result = convert_parquet_files_to_dfs(bucket_name="transform_bucket", client=mock_s3_client) # assert "dim_staff" in result + class TestUploadDfsToDatabase: - pass \ No newline at end of file + pass -- cgit v1.2.3