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/transform_lambda/dataframes.py | 228 +++++++++++++++++++++++++++++++ src/transform_lambda/transform_lambda.py | 217 +++++++++++++++++++++++++++++ 2 files changed, 445 insertions(+) create mode 100644 src/transform_lambda/dataframes.py create mode 100644 src/transform_lambda/transform_lambda.py (limited to 'src/transform_lambda') 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 From c6e711bd4196ba1c5b65218d347da1e7b98cac12 Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Wed, 28 Aug 2024 10:37:48 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 4651e2f according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/106 --- src/transform_lambda/transform_lambda.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) (limited to 'src/transform_lambda') diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py index c25ab39..8a2cae8 100644 --- a/src/transform_lambda/transform_lambda.py +++ b/src/transform_lambda/transform_lambda.py @@ -11,7 +11,6 @@ from pg8000.native import Connection, InterfaceError from datetime import datetime - class DBConnectionException(Exception): """Wraps pg8000.native Error or DatabaseError.""" @@ -115,13 +114,16 @@ def process_to_parquet_and_upload_to_s3( if table_name in existing_s3_files: status["not_uploaded"].append(table_name) else: -<<<<<<< HEAD:src/transform_lambda/transform_lambda.py + + +<< << << < HEAD: src/transform_lambda/transform_lambda.py 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") -======= + client.upload_file(f"{table_name}.parquet", + bucket, f"{table_name}.parquet") +== == == = parquet_buffer = io.BytesIO() # or engine="fastparquet" @@ -129,9 +131,10 @@ def process_to_parquet_and_upload_to_s3( parquet_buffer.seek(0) - client.upload_fileobj(parquet_buffer, bucket, f"{table_name}.parquet") + client.upload_fileobj(parquet_buffer, bucket, + f"{table_name}.parquet") ->>>>>>> 3f24ec753902feecec4c17e2877e19853bde1bb2:src/transform_lambda.py +>>>>>> > 3f24ec753902feecec4c17e2877e19853bde1bb2: src/transform_lambda.py status["uploaded"].append(table_name) for table_name, df in mutable_df_dict.items(): -- cgit v1.2.3 From 6c8567770042ad547366f0f02b091379a88d60d6 Mon Sep 17 00:00:00 2001 From: Alex Date: Wed, 28 Aug 2024 10:50:47 +0000 Subject: chore: get out of merge hell --- src/transform_lambda/transform_lambda.py | 21 ++------------------- 1 file changed, 2 insertions(+), 19 deletions(-) (limited to 'src/transform_lambda') diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py index 8a2cae8..02e9887 100644 --- a/src/transform_lambda/transform_lambda.py +++ b/src/transform_lambda/transform_lambda.py @@ -114,27 +114,12 @@ def process_to_parquet_and_upload_to_s3( if table_name in existing_s3_files: status["not_uploaded"].append(table_name) else: - - -<< << << < HEAD: src/transform_lambda/transform_lambda.py 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") -== == == = - parquet_buffer = io.BytesIO() - - # or engine="fastparquet" - df.to_parquet(parquet_buffer, engine="pyarrow") - - parquet_buffer.seek(0) - - client.upload_fileobj(parquet_buffer, bucket, - f"{table_name}.parquet") - ->>>>>> > 3f24ec753902feecec4c17e2877e19853bde1bb2: src/transform_lambda.py status["uploaded"].append(table_name) for table_name, df in mutable_df_dict.items(): @@ -205,12 +190,10 @@ def bucket_name(bucket_prefix, client=boto3.client("s3")): bucket["Name"] for bucket in response["Buckets"] if bucket_prefix in bucket["Name"] - ] -<<<<<<< HEAD:src/transform_lambda/transform_lambda.py -======= + ] + if not bucket_filter: raise ValueError(f"No bucket found with prefix: {bucket_prefix}") ->>>>>>> 3f24ec753902feecec4c17e2877e19853bde1bb2:src/transform_lambda.py return bucket_filter[0] -- cgit v1.2.3 From bf55c50ed6228eb1ca3b10e7280ed35944f7f42f Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Wed, 28 Aug 2024 10:51:00 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 6c85677 according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/106 --- src/transform_lambda/transform_lambda.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) (limited to 'src/transform_lambda') diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py index 02e9887..3dbb57b 100644 --- a/src/transform_lambda/transform_lambda.py +++ b/src/transform_lambda/transform_lambda.py @@ -118,8 +118,7 @@ def process_to_parquet_and_upload_to_s3( 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") + 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(): @@ -190,8 +189,8 @@ def bucket_name(bucket_prefix, client=boto3.client("s3")): bucket["Name"] for bucket in response["Buckets"] if bucket_prefix in bucket["Name"] - ] - + ] + if not bucket_filter: raise ValueError(f"No bucket found with prefix: {bucket_prefix}") -- 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(-) (limited to 'src/transform_lambda') 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 6235a2bb04b60d57a41196b07bbf0296920c6980 Mon Sep 17 00:00:00 2001 From: T-Aji Date: Wed, 28 Aug 2024 17:52:45 +0100 Subject: wip commit --- src/load_lambda.py | 174 +++++++++++++++++++------------ src/transform_lambda/dataframes.py | 8 +- src/transform_lambda/transform_lambda.py | 2 +- tests/test_transform_lambda.py | 2 +- 4 files changed, 115 insertions(+), 71 deletions(-) (limited to 'src/transform_lambda') 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(-) (limited to 'src/transform_lambda') 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 08:57:48 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 48e7dae according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/107 --- src/load_lambda.py | 78 +++++++++++++-------- src/transform_lambda/dataframes.py | 116 ++++++++++++++----------------- src/transform_lambda/transform_lambda.py | 6 +- 3 files changed, 105 insertions(+), 95 deletions(-) (limited to 'src/transform_lambda') 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,"<<