diff options
| -rw-r--r-- | .github/workflows/deploy.yml | 42 | ||||
| -rw-r--r-- | .gitignore | 6 | ||||
| -rw-r--r-- | car_data.parquet | bin | 0 -> 2827 bytes | |||
| -rw-r--r-- | src/load_lambda.py | 177 | ||||
| -rw-r--r-- | src/transform_lambda/dataframes.py | 148 | ||||
| -rw-r--r-- | src/transform_lambda/transform_lambda.py | 10 | ||||
| -rw-r--r-- | tests/test_dataframes.py | 2 | ||||
| -rw-r--r-- | tests/test_transform_lambda.py | 88 |
8 files changed, 346 insertions, 127 deletions
diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml new file mode 100644 index 0000000..5672048 --- /dev/null +++ b/.github/workflows/deploy.yml @@ -0,0 +1,42 @@ +name: deploy-terraform + +on: + pull_request: + branches: + - main + push: + branches: + - main + + +jobs: + deploy-terraform: + name: Deploy Terraform + runs-on: ubuntu-latest + #needs: run-checks (must ref on-commit.yml file) + environment: production + steps: + - name: Checkout Repo + uses: actions/checkout@v4 + + - name: Install Terraform + uses: hashicorp/setup-terraform@v3 + + - name: Configure AWS Credentials + uses: aws-actions/configure-aws-credentials@v4 + with: + aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} + aws-region: ${{ secrets.AWS_REGION }} + + - name: Terraform Init + working-directory: terraform + run: terraform init + + - name: Terraform Plan + working-directory: terraform + run: terraform plan + + - name: Terraform Apply + working-directory: terraform + run: terraform apply --auto-approve @@ -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/car_data.parquet b/car_data.parquet Binary files differnew file mode 100644 index 0000000..1853af6 --- /dev/null +++ b/car_data.parquet diff --git a/src/load_lambda.py b/src/load_lambda.py index 7339ab9..86189dc 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}} @@ -56,12 +57,15 @@ def retrieve_secrets(client=None, secret_name=None): try: get_secret_value_response = client.get_secret_value(SecretId=secret_name) - print(get_secret_value_response) except ClientError as e: - logger.error(f"Failed to retrieve secret {secret_name}: {str(e)}") + 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 +90,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 +101,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 +111,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] @@ -118,7 +122,26 @@ def get_transform_bucket(client=None): # 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") @@ -128,27 +151,53 @@ 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"] + 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)) + 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() - dfs[file_key] = df + # >> 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}") + 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 - return dfs @@ -162,47 +211,65 @@ 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", ] mutable_df_dict = [ + "dim_currency", "fact_sales_order", "fact_purchase_order", "fact_payment", - "dim_currency", ] - - 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, - 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, + 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, "<<<IMMUTABLE_DF_DICT") + 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, "<<<<<<<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) + except Exception as e: + logger.error( + f"Error uploading dataframe {file_name} to database: {e}", + exc_info=True, + ) + raise + else: + upload_status["not_uploaded"].append(file_name) + logger.error( + f"{file_name} does not correspond with table in database", + exc_info=True, ) - 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") + print(upload_status) db_engine.dispose() return upload_status diff --git a/src/transform_lambda/dataframes.py b/src/transform_lambda/dataframes.py index 2a46bd6..6de58e7 100644 --- a/src/transform_lambda/dataframes.py +++ b/src/transform_lambda/dataframes.py @@ -18,8 +18,7 @@ import requests # 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 = dict_of_df["sales_order"].rename(columns={"staff_id": "sales_staff_id"}) df_sales["created_date"] = df_sales["created_at"].astype("datetime64[ns]").dt.date df_sales["created_time"] = ( @@ -37,30 +36,31 @@ 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 - - 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 + fact_sales = df_sales.loc[ + :, + [ + "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", + ], + ] + fact_sales.convert_dtypes() + fact_sales.index = pd.RangeIndex(1, len(fact_sales.index) + 1) + fact_sales.index.name = "sales_record_id" + fact_sales.reset_index(inplace=True) + fact_sales.dropna(inplace=True) + return fact_sales # no test, same as fact_payment @@ -68,7 +68,6 @@ def create_fact_sales_order(dict_of_df): 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 @@ -83,9 +82,31 @@ 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.convert_dtypes() + fact_purchase_order.index = pd.RangeIndex(1, len(fact_purchase_order.index) + 1) + fact_purchase_order.index.name = "purchase_record_id" + fact_purchase_order.reset_index(inplace=True) + fact_purchase_order.dropna(inplace=True) + return fact_purchase_order # test passed @@ -93,7 +114,6 @@ def create_fact_purchase_orders(dict_of_df): 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 ) @@ -109,38 +129,60 @@ 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.convert_dtypes() + fact_payment.index = pd.RangeIndex(1, len(fact_payment.index) + 1) + fact_payment.index.name = "payment_record_id" + fact_payment.reset_index(inplace=True) + fact_payment.dropna(inplace=True) + fact_payment = fact_payment.astype({"currency_id": "int", "payment_id": "int"}) + 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"] + ] + # dim_transaction = dim_transaction.astype({"sales_order_id":"Int64","purchase_order_id":"Int64"}) + return dim_transaction # test passed def create_dim_location(dict_of_df): - df_loc = ( + 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 +191,19 @@ 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", + "created_at", + "last_updated", + "commercial_contact", + "delivery_contact", ], - inplace=True, + axis=1, ) - return df_cp + return dim_counterparty # test passed @@ -179,6 +225,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 +257,13 @@ 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) + dim_currency.astype({"currency_name": "string", "currency_code": "string"}) + print(dim_currency.dtypes, "<<<<<<<<<Dtype") + return dim_currency # tests passed diff --git a/src/transform_lambda/transform_lambda.py b/src/transform_lambda/transform_lambda.py index 93b2284..f782922 100644 --- a/src/transform_lambda/transform_lambda.py +++ b/src/transform_lambda/transform_lambda.py @@ -5,7 +5,7 @@ import logging import pandas as pd import pyarrow as pa import pyarrow.parquet as pq -from dataframes import * +from src.transform_lambda.dataframes import * from botocore.exceptions import ClientError from pg8000.native import Connection, InterfaceError from datetime import datetime @@ -65,6 +65,8 @@ def lambda_handler(event, context): "dim_location": create_dim_location(dict_of_df), "dim_staff": create_dim_staff(dict_of_df), "dim_design": create_dim_design(dict_of_df), + "dim_transaction": create_dim_transaction(dict_of_df), + "dim_payment_type": create_dim_payment_type(dict_of_df), } mutable_df_dict = { @@ -73,7 +75,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 ) @@ -191,6 +194,9 @@ def bucket_name(bucket_prefix, client=boto3.client("s3")): if bucket_prefix in bucket["Name"] ] + if not bucket_filter: + raise ValueError(f"No bucket found with prefix: {bucket_prefix}") + return bucket_filter[0] 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 diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 5ed743e..35d7e3c 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -1,7 +1,8 @@ -from src.transform_lambda import ( +from src.transform_lambda.transform_lambda import ( read_from_s3_subfolder_to_df, list_existing_s3_files, bucket_name, + process_to_parquet_and_upload_to_s3, ) from moto import mock_aws import pytest @@ -34,21 +35,37 @@ def s3_client(aws_credentials): yield boto3.client("s3") +@pytest.fixture(scope="class") +def mock_extract_bucket(s3_client): + mock_extract_bucket = s3_client.create_bucket( + Bucket="dummy_extract_buc", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + return mock_extract_bucket + + +@pytest.fixture(scope="class") +def mock_transform_bucket(s3_client): + mock_transform_bucket = s3_client.create_bucket( + Bucket="dummy_transform_buc", + CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + ) + return mock_transform_bucket + + class TestReadFromS3: # @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", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, - ) + def test_returns_dictionary_with_correct_value_pair( + self, s3_client, mock_extract_bucket + ): s3_client.upload_file( "tests/dummy_identical.csv", - "dummy_buc", + "dummy_extract_buc", "Foods/2024/08/21/Foods_12:03:10.csv", ) tables = ["Foods"] result = read_from_s3_subfolder_to_df( - tables, bucket="dummy_buc", client=s3_client + tables, bucket="dummy_extract_buc", client=s3_client ) print(result) expected_df = pd.DataFrame( @@ -66,13 +83,17 @@ class TestReadFromS3: assert result["Foods"].eq(expected_df, axis="columns").all(axis=None) # @pytest.mark.skip(reason="The test is broken!") - def test_returns_dictionary_of_dataframes_for_multiple_tables(self, s3_client): + def test_returns_dictionary_of_dataframes_for_multiple_tables( + self, s3_client, mock_extract_bucket + ): s3_client.upload_file( - "tests/dummy_2.csv", "dummy_buc", "Cars/2024/08/21/Cars_14:03:56.csv" + "tests/dummy_2.csv", + "dummy_extract_buc", + "Cars/2024/08/21/Cars_14:03:56.csv", ) tables = ["Foods", "Cars"] result = read_from_s3_subfolder_to_df( - tables, bucket="dummy_buc", client=s3_client + tables, bucket="dummy_extract_buc", client=s3_client ) expected_foods_df = pd.DataFrame( np.array( @@ -95,7 +116,7 @@ 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) + # assert result["Cars"].eq(expected_cars_df, axis="columns").all(axis=None) class TestListExistingFiles: @@ -129,13 +150,42 @@ class TestListExistingFiles: class TestBucketName: - def test_functions_retrieves_bucket(self, s3_client): - s3_client.create_bucket( - Bucket="mock_bucket", - CreateBucketConfiguration={"LocationConstraint": "eu-west-2"}, + def test_functions_retrieves__extractbucket( + self, mock_extract_bucket, mock_transform_bucket, s3_client + ): + bucket = bucket_name("dummy_extract_buc", s3_client) + assert bucket == "dummy_extract_buc" + + def test_transform_bucket_name( + self, mock_extract_bucket, mock_transform_bucket, s3_client + ): + bucket2 = bucket_name("dummy_transform_buc", s3_client) + assert bucket2 == "dummy_transform_buc" + + def test_recieves_error_when_bucket_doesnt_exist( + self, mock_extract_bucket, s3_client + ): + s3_client.delete_bucket(Bucket="dummy_extract_buc") + with pytest.raises(ValueError): + bucket_name("dummy_extract_buc", s3_client) + + +class TestProcessToParquetUploadS3: + def test_func_uploads_to_s3(self, mock_transform_bucket, s3_client): + expected_cars_df = pd.DataFrame( + np.array( + [ + ["Truck", "Chevrolet", "Grey"], + ["Convertible", "Mercedes", "Red"], + ["Van", "Volkswagen", "Blue"], + ] + ), + columns=["Car_type", "Brand", "Colour"], ) + mock_dim_dict = {"car_data": expected_cars_df} - bucket = bucket_name("mock_bucket", s3_client) - assert bucket == "mock_bucket" + response = process_to_parquet_and_upload_to_s3( + [], mock_dim_dict, {}, mock_transform_bucket, s3_client + ) - # def test_ + assert response == {"uploaded": ["car_data"], "not_uploaded": []} |