aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorEllie <167526066+ellsymonds@users.noreply.github.com>2024-08-27 11:08:06 +0100
committerGitHub <noreply@github.com>2024-08-27 11:08:06 +0100
commit182ba54c0c340a4819bb7400f6eb204e15364387 (patch)
treee3668c638914cc3efe003604c9d4ac0a5dc93439
parent69edb14dad584d45fa6a83a90c08292b84795507 (diff)
parentc610d3fc42a610ca5daff80606f8e67f9d1e20f2 (diff)
downloadde-project-bentley-182ba54c0c340a4819bb7400f6eb204e15364387.tar.gz
de-project-bentley-182ba54c0c340a4819bb7400f6eb204e15364387.zip
Merge branch 'development' into feature/load-lambda
-rw-r--r--requirements.txt3
-rw-r--r--src/dataframes.py229
-rw-r--r--src/transform_lambda.py189
-rw-r--r--tests/test_fact_sales_order.py246
-rw-r--r--tests/test_transform_lambda.py79
5 files changed, 733 insertions, 13 deletions
diff --git a/requirements.txt b/requirements.txt
index 614a0ab..763b95a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -31,4 +31,5 @@ xmltodict==0.13.0
s3fs
pandas
pyarrow
-SQLAlchemy \ No newline at end of file
+SQLAlchemy
+bs4
diff --git a/src/dataframes.py b/src/dataframes.py
new file mode 100644
index 0000000..ab53063
--- /dev/null
+++ b/src/dataframes.py
@@ -0,0 +1,229 @@
+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
+
+
+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"] = pd.to_datetime(df_sales["created_at"]).dt.date
+ df_sales["created_time"] = pd.to_datetime(df_sales["created_at"]).dt.time
+ df_sales["last_updated_date"] = pd.to_datetime(df_sales["last_updated"]).dt.date
+ df_sales["last_updated_time"] = pd.to_datetime(df_sales["last_updated"]).dt.time
+ fact_sales_order = df_sales.loc[
+ :,
+ [
+ "sales_record_id",
+ "sales_order_id",
+ "created_date",
+ "created_time",
+ "last_updated_date",
+ "last_updated_time",
+ "sales_staff_id",
+ "counterparty_id",
+ "units_sold",
+ "unit_price",
+ "currency_id",
+ "design_id",
+ "agreed_payment_date",
+ "agreed_delivery_date",
+ "agreed_delivery_location_id",
+ ],
+ ]
+ return fact_sales_order
+
+
+# fact_purchase_order from purchase_order
+
+
+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"].date()
+ df_po["created_time"] = df_po["created_at"].dt.time
+ df_po["last_updated_date"] = df_po["last_updated_at"].date()
+ df_po["last_updated_time"] = df_po["last_updated_at"].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.drop(labels=["created_at", "last_updated_at"], axis=1, inplace=True)
+ return df_po
+
+
+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"].date()
+ df_payment["created_time"] = df_payment["created_at"].time
+ df_payment["last_updated_date"] = df_payment["last_updated"].date()
+ df_payment["last_updated_time"] = df_payment["last_updated"].time
+ df_payment["payment_date"] = pd.to_datetime(
+ df_payment["payment_date"], format="%Y-%m-%d"
+ )
+ fact_payment = df_payment.loc[
+ :,
+ [
+ "payment_record_id",
+ "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",
+ ],
+ ]
+ 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
+
+
+# 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),
+ ]
+ date_col_names = [
+ col_name for col_name in list(fact_dfs[0].columns) if "date" in col_name
+ ]
+ list_of_date_columns = []
+ for df in fact_dfs:
+ for col in date_col_names:
+ list_of_date_columns.append(df[col])
+ sr_date = pd.array(pd.concat(list_of_date_columns), dtype="datetime64[ns]")
+ df_date = pd.DataFrame(data=sr_date, columns=["date_id"])
+ df_date.drop_duplicates(inplace=True)
+ df_date["year"] = df_date["date_id"].dt.year
+ df_date["month"] = df_date["date_id"].dt.month
+ df_date["day"] = df_date["date_id"].dt.day
+ df_date["day_of_week"] = df_date["date_id"].dt.dayofweek
+ df_date["day_name"] = df_date["date_id"].dt.day_name()
+ df_date["month_name"] = df_date["date_id"].dt.month_name()
+ df_date["quarter"] = df_date["date_id"].dt.quarter
+ return df_date
+
+
+# tests passed
+def scrape_currency_names():
+ response = requests.get("https://www.xe.com/currency/").content
+ soup = BeautifulSoup(response, "html.parser")
+ currency = [
+ item.text for item in soup.findAll("a", attrs={"class": "sc-299dec64-6 fZPTSw"})
+ ]
+ sr = pd.Series(currency)
+ df_cur = sr.str.split(pat=" - ", expand=True).rename(
+ {0: "currency_code", 1: "currency_name"}, axis=1
+ )
+ return df_cur
+
+
+# tests passed
+
+
+def create_dim_currency(dict_of_df, names=scrape_currency_names()):
+ df_cur = dict_of_df["currency"].drop(labels=["created_at", "last_updated"], axis=1)
+ dim_cur = pd.merge(
+ df_cur, names, left_on="currency_code", right_on="currency_code", how="inner"
+ )
+ return dim_cur
+
+
+# tests passed
+
+
+def create_dim_payment_type(dict_of_df):
+ df_payment_type = dict_of_df["payment_type"]
+ dim_payment_type = df_payment_type.loc[:, ["payment_type_id", "payment_type_name"]]
+ return dim_payment_type
+
+
+# tests passed
+
+
+def create_dim_design(dict_of_df):
+ df_design = dict_of_df["design"]
+ dim_design = df_design.loc[
+ :, ["design_id", "design_name", "file_name", "file_location"]
+ ]
+ return dim_design
+
+
+# tests passed
+
+
+def create_dim_staff(dict_of_df):
+ staff_department = pd.merge(
+ dict_of_df["staff"], dict_of_df["department"], on="department_id", how="left"
+ )
+ dim_staff = staff_department.loc[
+ :,
+ [
+ "staff_id",
+ "first_name",
+ "last_name",
+ "department_name",
+ "location",
+ "email_address",
+ ],
+ ]
+ return dim_staff
diff --git a/src/transform_lambda.py b/src/transform_lambda.py
index 9238180..2cd9272 100644
--- a/src/transform_lambda.py
+++ b/src/transform_lambda.py
@@ -1,16 +1,37 @@
import json
import boto3
import re
-import io
-from io import StringIO
+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
-def lambda_handler(event, context):
- pass
+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__)
-tables = [
+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",
@@ -25,6 +46,130 @@ tables = [
]
+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
+ client.upload_file(parquet_file, 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(parquet_file, 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:
@@ -35,3 +180,37 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")):
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 None
+
+ except ClientError as e:
+ logger.error(f"Error listing S3 objects: {e}")
+ raise e
+
+ return existing_files
+
+
+if __name__ == "__main__":
+ lambda_handler({}, "")
diff --git a/tests/test_fact_sales_order.py b/tests/test_fact_sales_order.py
new file mode 100644
index 0000000..a245379
--- /dev/null
+++ b/tests/test_fact_sales_order.py
@@ -0,0 +1,246 @@
+from src.dataframes import *
+import pandas as pd
+from unittest.mock import patch
+from datetime import datetime as dt
+
+
+class TestCreateDimDesign:
+ def test_dim_design_returns_dataframe(self):
+ d = {
+ "test": ["Hello", "Bye"],
+ "design_id": ["Hello", "Bye"],
+ "design_name": ["Hello", "Bye"],
+ "file_name": ["Hello", "Bye"],
+ "file_location": ["Hello", "Bye"],
+ "Hello": ["Hello", "Bye"],
+ }
+ test_df = {"design": pd.DataFrame(data=d)}
+ result = create_dim_design(test_df)
+ assert isinstance(result, pd.DataFrame)
+
+ def test_dim_design_returns_correct_columns_and_values(self):
+ d = {
+ "test": ["Hello", "Bye"],
+ "design_id": ["Hello", "Bye"],
+ "design_name": ["Hello", "Bye"],
+ "file_name": ["Hello", "Bye"],
+ "file_location": ["Hello", "Bye"],
+ "Hello": ["Hello", "Bye"],
+ }
+ test_df = {"design": pd.DataFrame(data=d)}
+ result = create_dim_design(test_df)
+ d2 = {
+ "design_id": ["Hello", "Bye"],
+ "design_name": ["Hello", "Bye"],
+ "file_name": ["Hello", "Bye"],
+ "file_location": ["Hello", "Bye"],
+ }
+ expected_df = pd.DataFrame(data=d2)
+ expected_result = expected_df.copy()
+ assert result.equals(expected_result)
+
+
+class TestCreateDimStaff:
+ def test_dim_staff_returns_dataframe(self):
+ d = {
+ "staff_id": ["Hello", "Bye"],
+ "first_name": ["Hello", "Bye"],
+ "last_name": ["Hello", "Bye"],
+ "department_id": ["Hello", "Bye"],
+ }
+ d2 = {
+ "department_name": ["Hello", "Bye"],
+ "location": ["Hello", "Bye"],
+ "email_address": ["Hello", "Bye"],
+ "department_id": ["Hello", "Bye"],
+ }
+ test_df = {"staff": pd.DataFrame(data=d), "department": pd.DataFrame(data=d2)}
+ result = create_dim_staff(test_df)
+ assert isinstance(result, pd.DataFrame)
+
+ def test_dim_staff_returns_correct_columns_and_values(self):
+ d = {
+ "staff_id": ["Hello", "Bye"],
+ "first_name": ["Hello", "Bye"],
+ "last_name": ["Hello", "Bye"],
+ "department_id": ["Hello", "Bye"],
+ }
+ d2 = {
+ "department_name": ["Hello", "Bye"],
+ "location": ["Hello", "Bye"],
+ "email_address": ["Hello", "Bye"],
+ "department_id": ["Hello", "Bye"],
+ }
+ test_df = {"staff": pd.DataFrame(data=d), "department": pd.DataFrame(data=d2)}
+ result = create_dim_staff(test_df)
+ expected_d = {
+ "staff_id": ["Hello", "Bye"],
+ "first_name": ["Hello", "Bye"],
+ "last_name": ["Hello", "Bye"],
+ "department_name": ["Hello", "Bye"],
+ "location": ["Hello", "Bye"],
+ "email_address": ["Hello", "Bye"],
+ }
+ expected_df = pd.DataFrame(data=expected_d)
+ expected_result = expected_df.copy()
+ assert result.equals(expected_result)
+
+
+class TestCreatePaymentType:
+ def test_create_dim_payment_type_returns_correct_columns_and_values(self):
+ d = {"payment_type_id": ["Hello", "Bye"], "payment_type_name": ["Hello", "Bye"]}
+ test_df = {"payment_type": pd.DataFrame(data=d)}
+ result = create_dim_payment_type(test_df)
+ expected_columns = ["payment_type_id", "payment_type_name"]
+ expected_d = {
+ "payment_type_id": ["Hello", "Bye"],
+ "payment_type_name": ["Hello", "Bye"],
+ }
+ expected_df = pd.DataFrame(data=expected_d)
+ assert isinstance(result, pd.DataFrame)
+ assert list(result.columns) == expected_columns
+ assert result.equals(expected_df)
+
+
+class TestCreateDimCounterparty:
+ def test_create_dim_counterparty_type_returns_correct_columns_and_object(self):
+ data_l = pd.DataFrame(
+ data={
+ "counterparty_id": ["Hello", "Bye"],
+ "counterparty_legal_name": ["Hello", "Bye"],
+ "commercial_contact": ["Hello", "Bye"],
+ "legal_address_id": ["bond street", "regent street"],
+ }
+ )
+ data_a = pd.DataFrame(
+ data={
+ "address_id": ["bond street", "regent street"],
+ "postcode": [98365, 93753],
+ }
+ )
+ test_df = {"address": data_a, "counterparty": data_l}
+ result = create_dim_counterparty(test_df)
+
+ expected_columns = [
+ "counterparty_id",
+ "counterparty_legal_name",
+ "commercial_contact",
+ "counterparty_legal_postcode",
+ ]
+ print(data_l)
+ print(data_a)
+ assert isinstance(result, pd.DataFrame)
+ assert list(result.columns) == expected_columns
+
+
+class TestCreateDimCurrency:
+ def test_dim_currency_returns_columns_and_values(self):
+ nones = [None, None, None]
+ d = {
+ "currency_id": [1, 2, 3],
+ "currency_code": ["USD", "EUR", "GBP"],
+ "created_at": nones,
+ "last_updated": nones,
+ }
+ test_df = {"currency": pd.DataFrame(data=d)}
+ scraper_output = pd.DataFrame(
+ {
+ "currency_code": ["RUS", "USD", "PHP", "GBP", "EUR"],
+ "currency_name": ["Rubble", "US Dollar", "Peso", "Pound", "Euro"],
+ }
+ )
+ result = create_dim_currency(test_df, names=scraper_output).sort_values(
+ by="currency_code", axis=0
+ )
+ expected_d = {
+ "currency_id": [1, 2, 3],
+ "currency_code": ["USD", "EUR", "GBP"],
+ "currency_name": ["US Dollar", "Euro", "Pound"],
+ }
+ expected_df = pd.DataFrame(data=expected_d).sort_values(
+ by="currency_code", axis=0
+ )
+ assert isinstance(result, pd.DataFrame)
+ assert result.equals(expected_df)
+
+ def test_scrape_currency_names_returns_dataframe_with_correct_collumns(self):
+ result = scrape_currency_names()
+ assert isinstance(result, pd.DataFrame)
+ assert list(result.columns) == ["currency_code", "currency_name"]
+
+
+class TestCreateDimDate:
+ def test_returns_required_columns(self):
+ df_one = pd.DataFrame(
+ data={
+ "updated_date": dt(2020, 5, 17),
+ "created_date": dt(2021, 5, 13),
+ "not_dat": None,
+ },
+ index=[0],
+ )
+ df_two = pd.DataFrame(
+ data={"updated_date": dt(2020, 5, 17), "created_date": dt(2021, 9, 13)},
+ index=[0],
+ )
+ df_three = pd.DataFrame(
+ data={"updated_date": dt(2022, 5, 17), "created_date": dt(2023, 5, 13)},
+ index=[0],
+ )
+ expected_df = pd.DataFrame(
+ data=[
+ [dt(2020, 5, 17), 2020, 5, 17, 6, "Sunday", "May", 2],
+ [dt(2021, 5, 13), 2021, 5, 13, 3, "Thursday", "May", 2],
+ [dt(2021, 9, 13), 2021, 9, 13, 0, "Monday", "September", 3],
+ [dt(2022, 5, 17), 2022, 5, 17, 1, "Tuesday", "May", 2],
+ [dt(2023, 5, 13), 2023, 5, 13, 5, "Saturday", "May", 2],
+ ],
+ columns=[
+ "date_id",
+ "year",
+ "month",
+ "day",
+ "day_of_week",
+ "day_name",
+ "month_name",
+ "quarter",
+ ],
+ )
+ with patch("src.dataframes.create_fact_payment") as mock_fp:
+ with patch("src.dataframes.create_fact_purchase_orders") as mock_fpo:
+ with patch("src.dataframes.create_fact_sales_order") as mock_fso:
+ mock_fp.return_value = df_one
+ mock_fpo.return_value = df_two
+ mock_fso.return_value = df_three
+ result = create_dim_date({"dum": 0})
+ result.reset_index(inplace=True, drop=True)
+ assert result.eq(expected_df, axis="columns").all(axis=None)
+
+
+class TestCreateDimLocation:
+ def test_returns_correct_columns_lo(self):
+ dict_df = {
+ "address": pd.DataFrame(
+ data=[["some_time", "some_other_time", 1, "SE18 9QO"]],
+ columns=["created_at", "last_updated", "address_id", "postal_code"],
+ )
+ }
+ result = create_dim_location(dict_df)
+ assert list(result.columns) == ["location_id", "postal_code"]
+
+
+class TestCreateDimTransaction:
+ def test_returns_correct_columns_tr(self):
+ dict_df = {
+ "transaction": pd.DataFrame(
+ data=[["some_time", "some_other_time", 1, "SE18 9QO"]],
+ columns=[
+ "created_at",
+ "last_updated",
+ "transaction_id",
+ "some_other_id",
+ ],
+ )
+ }
+ result = create_dim_transaction(dict_df)
+ assert list(result.columns) == ["transaction_id", "some_other_id"]
diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py
index 4c689f7..5ed743e 100644
--- a/tests/test_transform_lambda.py
+++ b/tests/test_transform_lambda.py
@@ -1,11 +1,23 @@
-from src.transform_lambda import read_from_s3_subfolder_to_df
+from src.transform_lambda import (
+ read_from_s3_subfolder_to_df,
+ list_existing_s3_files,
+ bucket_name,
+)
from moto import mock_aws
import pytest
import pandas as pd
import os
import boto3
+from botocore.exceptions import ClientError
import numpy as np
+# import caplog
+import logging
+
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
@pytest.fixture(scope="class")
def aws_credentials():
@@ -23,7 +35,7 @@ def s3_client(aws_credentials):
class TestReadFromS3:
- @pytest.mark.skip(reason="The test is broken!")
+ # @pytest.mark.skip(reason="The test is broken!")
def test_returns_dictionary_with_correct_value_pair(self, s3_client):
s3_client.create_bucket(
Bucket="dummy_buc",
@@ -40,15 +52,20 @@ class TestReadFromS3:
)
print(result)
expected_df = pd.DataFrame(
- np.array([["Vegetable", "Sour", "Green"], ["Berry", "Sweet", "Red"]]),
- columns=["Food_type", "Flavour", "Colour"],
+ np.array(
+ [
+ ["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"],
+ ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"],
+ ]
+ ),
+ columns=["Food_type", "Flavour", "Colour", "last_updated"],
)
assert isinstance(result, dict)
assert list(result.keys())[0] == "Foods"
assert isinstance(result["Foods"], pd.DataFrame)
assert result["Foods"].eq(expected_df, axis="columns").all(axis=None)
- @pytest.mark.skip(reason="The test is broken!")
+ # @pytest.mark.skip(reason="The test is broken!")
def test_returns_dictionary_of_dataframes_for_multiple_tables(self, s3_client):
s3_client.upload_file(
"tests/dummy_2.csv", "dummy_buc", "Cars/2024/08/21/Cars_14:03:56.csv"
@@ -58,8 +75,13 @@ class TestReadFromS3:
tables, bucket="dummy_buc", client=s3_client
)
expected_foods_df = pd.DataFrame(
- np.array([["Vegetable", "Sour", "Green"], ["Berry", "Sweet", "Red"]]),
- columns=["Food_type", "Flavour", "Colour"],
+ np.array(
+ [
+ ["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"],
+ ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"],
+ ]
+ ),
+ columns=["Food_type", "Flavour", "Colour", "last_updated"],
)
expected_cars_df = pd.DataFrame(
np.array(
@@ -74,3 +96,46 @@ class TestReadFromS3:
assert list(result.keys()) == tables
assert result["Foods"].eq(expected_foods_df, axis="columns").all(axis=None)
assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None)
+
+
+class TestListExistingFiles:
+ def test_functions_receives_error_if_no_bucket(self, s3_client, caplog):
+ caplog.set_level(logging.INFO)
+
+ with pytest.raises(ClientError):
+ list_existing_s3_files("rando_bucket", client=s3_client)
+
+ assert (
+ "Error listing S3 objects: An error occurred (NoSuchBucket) when calling the ListObjectsV2 operation: The specified bucket does not exist"
+ in caplog.text
+ )
+
+ def test_recieves_logger_error_if_no_files_listed(self, s3_client, caplog):
+ caplog.set_level(logging.INFO)
+
+ s3_client.create_bucket(
+ Bucket="mock_bucket",
+ CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
+ )
+ response = list_existing_s3_files("mock_bucket", client=s3_client)
+ assert "The bucket is empty" in caplog.text
+
+ def test_retrieves_existing_files(self, s3_client, caplog):
+ caplog.set_level(logging.INFO)
+
+ s3_client.upload_file("tests/dummy.txt", "mock_bucket", "dummy.txt")
+ result = list_existing_s3_files("mock_bucket", client=s3_client)
+ assert result == ["dummy.txt"]
+
+
+class TestBucketName:
+ def test_functions_retrieves_bucket(self, s3_client):
+ s3_client.create_bucket(
+ Bucket="mock_bucket",
+ CreateBucketConfiguration={"LocationConstraint": "eu-west-2"},
+ )
+
+ bucket = bucket_name("mock_bucket", s3_client)
+ assert bucket == "mock_bucket"
+
+ # def test_
git.ajschof.me — hosted by ajschofield — powered by cgit