diff options
Diffstat (limited to 'src/transform_lambda')
| -rw-r--r-- | src/transform_lambda/dataframes.py | 228 | ||||
| -rw-r--r-- | src/transform_lambda/transform_lambda.py | 217 |
2 files changed, 445 insertions, 0 deletions
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({}, "") |