From 7ff0716386cfb813034a3447949d0906ae6e09d1 Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 20 Aug 2024 22:59:33 +0100 Subject: ci: add dev-test.yml --- .github/workflows/dev-test.yml | 48 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) create mode 100644 .github/workflows/dev-test.yml diff --git a/.github/workflows/dev-test.yml b/.github/workflows/dev-test.yml new file mode 100644 index 0000000..ebdad5f --- /dev/null +++ b/.github/workflows/dev-test.yml @@ -0,0 +1,48 @@ +name: Development CI + +on: + pull_request: + branches: + - development + push: + branches: + - development + +jobs: + validate-and-test: + name: Validate Terraform and Run Tests + runs-on: ubuntu-latest + steps: + - name: Checkout Repo + uses: actions/checkout@v4 + + - name: Install Terraform + uses: hashicorp/setup-terraform@v3 + + - name: Terraform Init + working-directory: terraform + run: terraform init -backend=false + + - name: Terraform Validate + working-directory: terraform + run: terraform validate + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install Python dependencies + run: | + python -m pip install --upgrade pip + pip install pytest pytest-testdox + pip install -r requirements.txt + + - name: Run pytest + run: pytest tests/ -vvrP --testdox + continue-on-error: true + id: pytest + + - name: Check on failures + if: steps.pytest.outcome == 'failure' + run: exit 1 -- cgit v1.2.3 From 95e5e49aa544ec2bda244a1225a2a467983db22a Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 20 Aug 2024 23:04:55 +0100 Subject: ci: update dev-test.yml --- .github/workflows/dev-test.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/dev-test.yml b/.github/workflows/dev-test.yml index ebdad5f..a1e64b2 100644 --- a/.github/workflows/dev-test.yml +++ b/.github/workflows/dev-test.yml @@ -12,6 +12,7 @@ jobs: validate-and-test: name: Validate Terraform and Run Tests runs-on: ubuntu-latest + environment: testing steps: - name: Checkout Repo uses: actions/checkout@v4 -- cgit v1.2.3 From c600a7694f770954e4c8b836de5640024d61c4e6 Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 20 Aug 2024 23:07:55 +0100 Subject: ci: rm dev-test.yml It's in the wrong branch... --- .github/workflows/dev-test.yml | 49 ------------------------------------------ 1 file changed, 49 deletions(-) delete mode 100644 .github/workflows/dev-test.yml diff --git a/.github/workflows/dev-test.yml b/.github/workflows/dev-test.yml deleted file mode 100644 index a1e64b2..0000000 --- a/.github/workflows/dev-test.yml +++ /dev/null @@ -1,49 +0,0 @@ -name: Development CI - -on: - pull_request: - branches: - - development - push: - branches: - - development - -jobs: - validate-and-test: - name: Validate Terraform and Run Tests - runs-on: ubuntu-latest - environment: testing - steps: - - name: Checkout Repo - uses: actions/checkout@v4 - - - name: Install Terraform - uses: hashicorp/setup-terraform@v3 - - - name: Terraform Init - working-directory: terraform - run: terraform init -backend=false - - - name: Terraform Validate - working-directory: terraform - run: terraform validate - - - name: Set up Python - uses: actions/setup-python@v5 - with: - python-version: '3.11' - - - name: Install Python dependencies - run: | - python -m pip install --upgrade pip - pip install pytest pytest-testdox - pip install -r requirements.txt - - - name: Run pytest - run: pytest tests/ -vvrP --testdox - continue-on-error: true - id: pytest - - - name: Check on failures - if: steps.pytest.outcome == 'failure' - run: exit 1 -- cgit v1.2.3 From da3d85dd2dc515226d16992c5f63b2a8b02a0a38 Mon Sep 17 00:00:00 2001 From: Ellie Date: Wed, 21 Aug 2024 13:41:01 +0100 Subject: add dim tables: design, staff, currency, location (wip) --- src/fact-sales-order.py | 54 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) create mode 100644 src/fact-sales-order.py diff --git a/src/fact-sales-order.py b/src/fact-sales-order.py new file mode 100644 index 0000000..a143889 --- /dev/null +++ b/src/fact-sales-order.py @@ -0,0 +1,54 @@ +import pandas as pd +from src.transform_lambda import get_dataframes + +dict_of_df = get_dataframes() # {"design": "design dataframe", "address": "address dataframe", ....} + + +# iterates through each dataframe in the list of dataframes and assigns them to a variable +df_design = dict_of_df[design] +df_currency = dict_of_df[currency] +df_address = dict_of_df[address] +df_staff = dict_of_df[staff] +df_department = dict_of_df[department] +df_counterparty = dict_of_df[counterparty] + + +# creates the dim_design dataframe +dim_design = df_design["design_id", "design_name", "file_name", "file_location"] + +# creates the dim_staff dataframe +staff_department = pd.merge(df_staff, df_department, on='department_id', how="outer") +dim_staff = staff_department['staff_id', 'first_name', 'last_name', 'department_name', 'location', 'email_address'] + +# creates the dim_currency dataframe +# currency names currently hardcoded and not taken from database, is this viable/how else to do this? +d = {"currency_id": [1, 2, 3], "currency_code": ["GBP", "USD", "EUR"], "currency_name": ["Pound", "US Dollar", "Euro"]} +currency_names = pd.DataFrame(data=d) +join_currency = pd.merge(df_currency, currency_names, on="currency_name", how="outer") +dim_currency = join_currency["currency_id", "currency_code", "currency_name"] + +# creates the dim_location dataframe +# need to change address id to location id +"dim_location dataframe: (location_id, address_line_1, address_line_2, district, city, postal code, country, phone)" +dim_location = df_address["address_id", "address_line_1", "address_line_2", "district", "city", "postal_code" "country", "phone"] + + + + + + + + + +# creates the dim_counterparty dataframe +# counterparty_address = pd.merge(df_counterparty, df_address, left_on="legal_address_id", right_on='address_id', how="outer") + +# dim_counterparty = df_counterparty["counterparty_id", "counterparty_legal_name", "counterparty_legal_address_line_1", +# "counterparty_legal_address_line_2", "counterparty_legal_district", "counterpart_legal_city", +# "counterparty_legal_postal_code", "counterparty_legal_country", "counterparty_legal_phone_number"] + + +# TO DO: +# dim_location +# dim_date +# fact_sales_order \ No newline at end of file -- cgit v1.2.3 From ccedcc10ed533688188a82d2fd364032a326941f Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 20 Aug 2024 22:59:33 +0100 Subject: ci: add dev-test.yml --- .github/workflows/dev-test.yml | 48 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) create mode 100644 .github/workflows/dev-test.yml diff --git a/.github/workflows/dev-test.yml b/.github/workflows/dev-test.yml new file mode 100644 index 0000000..ebdad5f --- /dev/null +++ b/.github/workflows/dev-test.yml @@ -0,0 +1,48 @@ +name: Development CI + +on: + pull_request: + branches: + - development + push: + branches: + - development + +jobs: + validate-and-test: + name: Validate Terraform and Run Tests + runs-on: ubuntu-latest + steps: + - name: Checkout Repo + uses: actions/checkout@v4 + + - name: Install Terraform + uses: hashicorp/setup-terraform@v3 + + - name: Terraform Init + working-directory: terraform + run: terraform init -backend=false + + - name: Terraform Validate + working-directory: terraform + run: terraform validate + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install Python dependencies + run: | + python -m pip install --upgrade pip + pip install pytest pytest-testdox + pip install -r requirements.txt + + - name: Run pytest + run: pytest tests/ -vvrP --testdox + continue-on-error: true + id: pytest + + - name: Check on failures + if: steps.pytest.outcome == 'failure' + run: exit 1 -- cgit v1.2.3 From 24ad8521b88c6a9b43c74d69443895872b8917ec Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 20 Aug 2024 23:04:55 +0100 Subject: ci: update dev-test.yml --- .github/workflows/dev-test.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/dev-test.yml b/.github/workflows/dev-test.yml index ebdad5f..a1e64b2 100644 --- a/.github/workflows/dev-test.yml +++ b/.github/workflows/dev-test.yml @@ -12,6 +12,7 @@ jobs: validate-and-test: name: Validate Terraform and Run Tests runs-on: ubuntu-latest + environment: testing steps: - name: Checkout Repo uses: actions/checkout@v4 -- cgit v1.2.3 From 095acc642a5abbf79209040aa2ac3d413a4ff49a Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 20 Aug 2024 23:07:55 +0100 Subject: ci: rm dev-test.yml It's in the wrong branch... --- .github/workflows/dev-test.yml | 49 ------------------------------------------ 1 file changed, 49 deletions(-) delete mode 100644 .github/workflows/dev-test.yml diff --git a/.github/workflows/dev-test.yml b/.github/workflows/dev-test.yml deleted file mode 100644 index a1e64b2..0000000 --- a/.github/workflows/dev-test.yml +++ /dev/null @@ -1,49 +0,0 @@ -name: Development CI - -on: - pull_request: - branches: - - development - push: - branches: - - development - -jobs: - validate-and-test: - name: Validate Terraform and Run Tests - runs-on: ubuntu-latest - environment: testing - steps: - - name: Checkout Repo - uses: actions/checkout@v4 - - - name: Install Terraform - uses: hashicorp/setup-terraform@v3 - - - name: Terraform Init - working-directory: terraform - run: terraform init -backend=false - - - name: Terraform Validate - working-directory: terraform - run: terraform validate - - - name: Set up Python - uses: actions/setup-python@v5 - with: - python-version: '3.11' - - - name: Install Python dependencies - run: | - python -m pip install --upgrade pip - pip install pytest pytest-testdox - pip install -r requirements.txt - - - name: Run pytest - run: pytest tests/ -vvrP --testdox - continue-on-error: true - id: pytest - - - name: Check on failures - if: steps.pytest.outcome == 'failure' - run: exit 1 -- cgit v1.2.3 From 4dc7b885950d7c352c53cdd31ac7bb0e905304dd Mon Sep 17 00:00:00 2001 From: Ellie Date: Wed, 21 Aug 2024 13:41:01 +0100 Subject: add dim tables: design, staff, currency, location (wip) --- src/fact-sales-order.py | 54 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) create mode 100644 src/fact-sales-order.py diff --git a/src/fact-sales-order.py b/src/fact-sales-order.py new file mode 100644 index 0000000..a143889 --- /dev/null +++ b/src/fact-sales-order.py @@ -0,0 +1,54 @@ +import pandas as pd +from src.transform_lambda import get_dataframes + +dict_of_df = get_dataframes() # {"design": "design dataframe", "address": "address dataframe", ....} + + +# iterates through each dataframe in the list of dataframes and assigns them to a variable +df_design = dict_of_df[design] +df_currency = dict_of_df[currency] +df_address = dict_of_df[address] +df_staff = dict_of_df[staff] +df_department = dict_of_df[department] +df_counterparty = dict_of_df[counterparty] + + +# creates the dim_design dataframe +dim_design = df_design["design_id", "design_name", "file_name", "file_location"] + +# creates the dim_staff dataframe +staff_department = pd.merge(df_staff, df_department, on='department_id', how="outer") +dim_staff = staff_department['staff_id', 'first_name', 'last_name', 'department_name', 'location', 'email_address'] + +# creates the dim_currency dataframe +# currency names currently hardcoded and not taken from database, is this viable/how else to do this? +d = {"currency_id": [1, 2, 3], "currency_code": ["GBP", "USD", "EUR"], "currency_name": ["Pound", "US Dollar", "Euro"]} +currency_names = pd.DataFrame(data=d) +join_currency = pd.merge(df_currency, currency_names, on="currency_name", how="outer") +dim_currency = join_currency["currency_id", "currency_code", "currency_name"] + +# creates the dim_location dataframe +# need to change address id to location id +"dim_location dataframe: (location_id, address_line_1, address_line_2, district, city, postal code, country, phone)" +dim_location = df_address["address_id", "address_line_1", "address_line_2", "district", "city", "postal_code" "country", "phone"] + + + + + + + + + +# creates the dim_counterparty dataframe +# counterparty_address = pd.merge(df_counterparty, df_address, left_on="legal_address_id", right_on='address_id', how="outer") + +# dim_counterparty = df_counterparty["counterparty_id", "counterparty_legal_name", "counterparty_legal_address_line_1", +# "counterparty_legal_address_line_2", "counterparty_legal_district", "counterpart_legal_city", +# "counterparty_legal_postal_code", "counterparty_legal_country", "counterparty_legal_phone_number"] + + +# TO DO: +# dim_location +# dim_date +# fact_sales_order \ No newline at end of file -- cgit v1.2.3 From 74be9f231ad560eed8630125045532b5975553dc Mon Sep 17 00:00:00 2001 From: T-Aji Date: Wed, 21 Aug 2024 15:58:45 +0100 Subject: 5 dim tables created --- src/fact-sales-order.py | 48 +++++++++++++++++++++++++++++++++--------------- 1 file changed, 33 insertions(+), 15 deletions(-) diff --git a/src/fact-sales-order.py b/src/fact-sales-order.py index a143889..30c958f 100644 --- a/src/fact-sales-order.py +++ b/src/fact-sales-order.py @@ -11,7 +11,7 @@ df_address = dict_of_df[address] df_staff = dict_of_df[staff] df_department = dict_of_df[department] df_counterparty = dict_of_df[counterparty] - +df_sales = dict_of_df[sales] # creates the dim_design dataframe dim_design = df_design["design_id", "design_name", "file_name", "file_location"] @@ -27,28 +27,46 @@ currency_names = pd.DataFrame(data=d) join_currency = pd.merge(df_currency, currency_names, on="currency_name", how="outer") dim_currency = join_currency["currency_id", "currency_code", "currency_name"] -# creates the dim_location dataframe -# need to change address id to location id -"dim_location dataframe: (location_id, address_line_1, address_line_2, district, city, postal code, country, phone)" -dim_location = df_address["address_id", "address_line_1", "address_line_2", "district", "city", "postal_code" "country", "phone"] - - - +# Using .map to add currency_name column and link it to the currency code +# dim_currency = df_currency["currency_id", "currency_code"] +# mappings = { +# "GBP": "Pound", +# "USD": "US Dollar", +# "EUR": "Euro" +# } +# dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) +# creates the dim_location dataframe +# need to change address id to location id +"dim_location dataframe: (location_id, address_line_1, address_line_2, district, city, postal code, country, phone)" +df_address.rename(columns={"address_id": "location_id"}) +dim_location = df_address["location_id", "address_line_1", "address_line_2", "district", "city", "postal_code" "country", "phone"] +# creates the dim_counterparty dataframe +counterparty_address = pd.merge(df_counterparty, df_address, left_on="legal_address_id", right_on='address_id', how="outer") +counterparty_address.rename(columns={"address_line_1": "counterparty_legal_address_line_1", "address_line_2": "counterparty_legal_address_line_2", + "district": "counterparty_legal_district", "city": "counterparty_legal_city", "postal_code": "counterparty_postal_code", + "country": "counterparty_legal_country", "phone": "counterparty_legal_phone_number"}) +dim_counterparty = df_counterparty["counterparty_id", "counterparty_legal_name", "counterparty_legal_address_line_1", + "counterparty_legal_address_line_2", "counterparty_legal_district", "counterpart_legal_city", + "counterparty_legal_postal_code", "counterparty_legal_country", "counterparty_legal_phone_number"] -# creates the dim_counterparty dataframe -# counterparty_address = pd.merge(df_counterparty, df_address, left_on="legal_address_id", right_on='address_id', how="outer") +# creates the dim_date dataframe +df_sales = df_sales["agreed_delivery_date"] +df_sales["agreed_delivery_date"] = pd.to_datetime["agreed_delivery_date"] +df_sales["year"] = df_sales["agreed_delivery_date"].dt.year +df_sales["month"] = df_sales["agreed_delivery_date"].dt.month +df_sales["day"] = df_sales["agreed_delivery_date"].dt.day +df_sales["day_of_week"] = df_sales["agreed_delivery_date"].dt.dayofweek +df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() +df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() +df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() -# dim_counterparty = df_counterparty["counterparty_id", "counterparty_legal_name", "counterparty_legal_address_line_1", -# "counterparty_legal_address_line_2", "counterparty_legal_district", "counterpart_legal_city", -# "counterparty_legal_postal_code", "counterparty_legal_country", "counterparty_legal_phone_number"] +dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() # TO DO: -# dim_location -# dim_date # fact_sales_order \ No newline at end of file -- cgit v1.2.3 From 0c02bd3636ed8815aadf73685c20f8c76a073c99 Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Wed, 21 Aug 2024 15:09:58 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 20a3bd8 according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/85 --- src/fact-sales-order.py | 86 ++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 68 insertions(+), 18 deletions(-) diff --git a/src/fact-sales-order.py b/src/fact-sales-order.py index 30c958f..399e435 100644 --- a/src/fact-sales-order.py +++ b/src/fact-sales-order.py @@ -1,7 +1,8 @@ import pandas as pd from src.transform_lambda import get_dataframes -dict_of_df = get_dataframes() # {"design": "design dataframe", "address": "address dataframe", ....} +# {"design": "design dataframe", "address": "address dataframe", ....} +dict_of_df = get_dataframes() # iterates through each dataframe in the list of dataframes and assigns them to a variable @@ -17,12 +18,23 @@ df_sales = dict_of_df[sales] dim_design = df_design["design_id", "design_name", "file_name", "file_location"] # creates the dim_staff dataframe -staff_department = pd.merge(df_staff, df_department, on='department_id', how="outer") -dim_staff = staff_department['staff_id', 'first_name', 'last_name', 'department_name', 'location', 'email_address'] +staff_department = pd.merge(df_staff, df_department, on="department_id", how="outer") +dim_staff = staff_department[ + "staff_id", + "first_name", + "last_name", + "department_name", + "location", + "email_address", +] # creates the dim_currency dataframe -# currency names currently hardcoded and not taken from database, is this viable/how else to do this? -d = {"currency_id": [1, 2, 3], "currency_code": ["GBP", "USD", "EUR"], "currency_name": ["Pound", "US Dollar", "Euro"]} +# currency names currently hardcoded and not taken from database, is this viable/how else to do this? +d = { + "currency_id": [1, 2, 3], + "currency_code": ["GBP", "USD", "EUR"], + "currency_name": ["Pound", "US Dollar", "Euro"], +} currency_names = pd.DataFrame(data=d) join_currency = pd.merge(df_currency, currency_names, on="currency_name", how="outer") dim_currency = join_currency["currency_id", "currency_code", "currency_name"] @@ -37,22 +49,51 @@ dim_currency = join_currency["currency_id", "currency_code", "currency_name"] # dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) - # creates the dim_location dataframe -# need to change address id to location id +# need to change address id to location id "dim_location dataframe: (location_id, address_line_1, address_line_2, district, city, postal code, country, phone)" df_address.rename(columns={"address_id": "location_id"}) -dim_location = df_address["location_id", "address_line_1", "address_line_2", "district", "city", "postal_code" "country", "phone"] +dim_location = df_address[ + "location_id", + "address_line_1", + "address_line_2", + "district", + "city", + "postal_code" "country", + "phone", +] # creates the dim_counterparty dataframe -counterparty_address = pd.merge(df_counterparty, df_address, left_on="legal_address_id", right_on='address_id', how="outer") -counterparty_address.rename(columns={"address_line_1": "counterparty_legal_address_line_1", "address_line_2": "counterparty_legal_address_line_2", - "district": "counterparty_legal_district", "city": "counterparty_legal_city", "postal_code": "counterparty_postal_code", - "country": "counterparty_legal_country", "phone": "counterparty_legal_phone_number"}) - -dim_counterparty = df_counterparty["counterparty_id", "counterparty_legal_name", "counterparty_legal_address_line_1", - "counterparty_legal_address_line_2", "counterparty_legal_district", "counterpart_legal_city", - "counterparty_legal_postal_code", "counterparty_legal_country", "counterparty_legal_phone_number"] +counterparty_address = pd.merge( + df_counterparty, + df_address, + left_on="legal_address_id", + right_on="address_id", + how="outer", +) +counterparty_address.rename( + columns={ + "address_line_1": "counterparty_legal_address_line_1", + "address_line_2": "counterparty_legal_address_line_2", + "district": "counterparty_legal_district", + "city": "counterparty_legal_city", + "postal_code": "counterparty_postal_code", + "country": "counterparty_legal_country", + "phone": "counterparty_legal_phone_number", + } +) + +dim_counterparty = df_counterparty[ + "counterparty_id", + "counterparty_legal_name", + "counterparty_legal_address_line_1", + "counterparty_legal_address_line_2", + "counterparty_legal_district", + "counterpart_legal_city", + "counterparty_legal_postal_code", + "counterparty_legal_country", + "counterparty_legal_phone_number", +] # creates the dim_date dataframe df_sales = df_sales["agreed_delivery_date"] @@ -65,8 +106,17 @@ df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() -dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() +dim_date = [ + "date_id", + "year", + "month", + "day", + "day_of_week", + "day_name", + "month_name", + "quarter", +] # series.dt.quarter() # TO DO: -# fact_sales_order \ No newline at end of file +# fact_sales_order -- cgit v1.2.3 From 5b2b4864eae129e112e70d093eb66498d7de401e Mon Sep 17 00:00:00 2001 From: lian-manonog Date: Wed, 21 Aug 2024 17:11:57 +0100 Subject: wip: fact_purchase_order schema --- src/fact-purchase-table.py | 34 ++++++++++++++++++++++++++++++++++ src/fact-sales-order.py | 2 +- src/transform_lambda.py | 4 ++-- 3 files changed, 37 insertions(+), 3 deletions(-) create mode 100644 src/fact-purchase-table.py diff --git a/src/fact-purchase-table.py b/src/fact-purchase-table.py new file mode 100644 index 0000000..53c0148 --- /dev/null +++ b/src/fact-purchase-table.py @@ -0,0 +1,34 @@ +from src.transform_lambda import read_from_s3_subfolder_to_df, tables +from src.extract_lambda import extract_bucket +import json +import boto3 +import re +import pandas as pd + + +dict_of_df = read_from_s3_subfolder_to_df(tables, extract_bucket(), client=boto3.client("s3")) + + +# iterates through each dataframe in the list of dataframes and assigns them to a variable +df_staff = dict_of_df['staff'] ##no change +df_currency = dict_of_df['currency'] ##scraping API +df_counterparty = dict_of_df['counterparty'] +df_address = dict_of_df['address'] +df_department = dict_of_df['department'] +df_purchase_order = dict_of_df['purchase_order'] + +## dim_staff table is the same across the schemas (no change) + +## dim_counterparty table + +## dim_location df_currency --> drops 2 columns +dim_location = df_address.drop(labels=['created_at', 'last_updated'], axis=1).rename(columns={'address_id': 'location_id'}) + +## dim_counterparty +df_prefixed_address = df_address.add_prefix('counterparty_legal_', axis=1) +pd.merge(df_counterparty, + df_prefixed_address, + left_on="legal_address_id", + right_on="address_id", + how="outer") + diff --git a/src/fact-sales-order.py b/src/fact-sales-order.py index 399e435..57e2e84 100644 --- a/src/fact-sales-order.py +++ b/src/fact-sales-order.py @@ -69,7 +69,7 @@ counterparty_address = pd.merge( df_address, left_on="legal_address_id", right_on="address_id", - how="outer", + how="outer" ) counterparty_address.rename( columns={ diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 9238180..920a24f 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -1,8 +1,6 @@ import json import boto3 import re -import io -from io import StringIO import pandas as pd @@ -35,3 +33,5 @@ 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 + + -- cgit v1.2.3 From 956bc9223a584c9cb687277f9000967f9b3ddc6b Mon Sep 17 00:00:00 2001 From: T-Aji Date: Wed, 21 Aug 2024 20:04:13 +0100 Subject: began dim_date df --- src/fact-sales-order.py | 35 +++++++++++++++++------------------ 1 file changed, 17 insertions(+), 18 deletions(-) diff --git a/src/fact-sales-order.py b/src/fact-sales-order.py index 30c958f..ef18f02 100644 --- a/src/fact-sales-order.py +++ b/src/fact-sales-order.py @@ -14,27 +14,21 @@ df_counterparty = dict_of_df[counterparty] df_sales = dict_of_df[sales] # creates the dim_design dataframe -dim_design = df_design["design_id", "design_name", "file_name", "file_location"] +dim_design = df_design.loc[:, "design_id", "design_name", "file_name", "file_location"] # creates the dim_staff dataframe staff_department = pd.merge(df_staff, df_department, on='department_id', how="outer") -dim_staff = staff_department['staff_id', 'first_name', 'last_name', 'department_name', 'location', 'email_address'] +dim_staff = staff_department.loc[:, 'staff_id', 'first_name', 'last_name', 'department_name', 'location', 'email_address'] # creates the dim_currency dataframe -# currency names currently hardcoded and not taken from database, is this viable/how else to do this? -d = {"currency_id": [1, 2, 3], "currency_code": ["GBP", "USD", "EUR"], "currency_name": ["Pound", "US Dollar", "Euro"]} -currency_names = pd.DataFrame(data=d) -join_currency = pd.merge(df_currency, currency_names, on="currency_name", how="outer") -dim_currency = join_currency["currency_id", "currency_code", "currency_name"] - # Using .map to add currency_name column and link it to the currency code -# dim_currency = df_currency["currency_id", "currency_code"] -# mappings = { -# "GBP": "Pound", -# "USD": "US Dollar", -# "EUR": "Euro" -# } -# dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) +dim_currency = df_currency.loc[:, "currency_id", "currency_code"] +mappings = { + "GBP": "Pound", + "USD": "US Dollar", + "EUR": "Euro" +} +dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) @@ -42,7 +36,7 @@ dim_currency = join_currency["currency_id", "currency_code", "currency_name"] # need to change address id to location id "dim_location dataframe: (location_id, address_line_1, address_line_2, district, city, postal code, country, phone)" df_address.rename(columns={"address_id": "location_id"}) -dim_location = df_address["location_id", "address_line_1", "address_line_2", "district", "city", "postal_code" "country", "phone"] +dim_location = df_address.loc[:, "location_id", "address_line_1", "address_line_2", "district", "city", "postal_code" "country", "phone"] # creates the dim_counterparty dataframe counterparty_address = pd.merge(df_counterparty, df_address, left_on="legal_address_id", right_on='address_id', how="outer") @@ -50,12 +44,12 @@ counterparty_address.rename(columns={"address_line_1": "counterparty_legal_addre "district": "counterparty_legal_district", "city": "counterparty_legal_city", "postal_code": "counterparty_postal_code", "country": "counterparty_legal_country", "phone": "counterparty_legal_phone_number"}) -dim_counterparty = df_counterparty["counterparty_id", "counterparty_legal_name", "counterparty_legal_address_line_1", +dim_counterparty = df_counterparty.loc[:, "counterparty_id", "counterparty_legal_name", "counterparty_legal_address_line_1", "counterparty_legal_address_line_2", "counterparty_legal_district", "counterpart_legal_city", "counterparty_legal_postal_code", "counterparty_legal_country", "counterparty_legal_phone_number"] # creates the dim_date dataframe -df_sales = df_sales["agreed_delivery_date"] +df_sales = df_sales.loc[:, "agreed_delivery_date"] df_sales["agreed_delivery_date"] = pd.to_datetime["agreed_delivery_date"] df_sales["year"] = df_sales["agreed_delivery_date"].dt.year df_sales["month"] = df_sales["agreed_delivery_date"].dt.month @@ -65,6 +59,11 @@ df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() +# repeat ln 52 - 60 for each column +# merge dataframes into one dataframe +# remove duplicates + + dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() -- cgit v1.2.3 From c5338ebb198a79604e36d65de39e28baf54f0ecd Mon Sep 17 00:00:00 2001 From: T-Aji Date: Thu, 22 Aug 2024 10:29:34 +0100 Subject: refactor df creation into func --- src/fact-sales-order.py | 104 ++++++++++++++++-------------------------------- 1 file changed, 34 insertions(+), 70 deletions(-) diff --git a/src/fact-sales-order.py b/src/fact-sales-order.py index 870f660..7921047 100644 --- a/src/fact-sales-order.py +++ b/src/fact-sales-order.py @@ -1,86 +1,50 @@ import pandas as pd -from src.transform_lambda import get_dataframes -# {"design": "design dataframe", "address": "address dataframe", ....} -dict_of_df = get_dataframes() +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 -# iterates through each dataframe in the list of dataframes and assigns them to a variable -df_design = dict_of_df[design] -df_currency = dict_of_df[currency] -df_address = dict_of_df[address] -df_staff = dict_of_df[staff] -df_department = dict_of_df[department] -df_counterparty = dict_of_df[counterparty] -df_sales = dict_of_df[sales] +def create_dim_staff(dict_of_df): + staff_department = pd.merge(dict_of_df["staff"], dict_of_df["department"], on='department_id', how="outer") + dim_staff = staff_department.loc[:, ['staff_id', 'first_name', 'last_name', 'department_name', 'location', 'email_address']] + return dim_staff -# creates the dim_design dataframe -dim_design = df_design.loc[:, "design_id", "design_name", "file_name", "file_location"] - -# creates the dim_staff dataframe -staff_department = pd.merge(df_staff, df_department, on='department_id', how="outer") -dim_staff = staff_department.loc[:, 'staff_id', 'first_name', 'last_name', 'department_name', 'location', 'email_address'] - -# creates the dim_currency dataframe -# Using .map to add currency_name column and link it to the currency code -dim_currency = df_currency.loc[:, "currency_id", "currency_code"] -mappings = { - "GBP": "Pound", - "USD": "US Dollar", - "EUR": "Euro" -} -dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) - - - -# creates the dim_location dataframe -# need to change address id to location id -"dim_location dataframe: (location_id, address_line_1, address_line_2, district, city, postal code, country, phone)" -df_address.rename(columns={"address_id": "location_id"}) -dim_location = df_address.loc[:, "location_id", "address_line_1", "address_line_2", "district", "city", "postal_code" "country", "phone"] - -# creates the dim_counterparty dataframe -counterparty_address = pd.merge( - df_counterparty, - df_address, - left_on="legal_address_id", - right_on="address_id", - how="outer" -) -counterparty_address.rename( - columns={ - "address_line_1": "counterparty_legal_address_line_1", - "address_line_2": "counterparty_legal_address_line_2", - "district": "counterparty_legal_district", - "city": "counterparty_legal_city", - "postal_code": "counterparty_postal_code", - "country": "counterparty_legal_country", - "phone": "counterparty_legal_phone_number", +def create_dim_currency(dict_of_df): + df_currency = dict_of_df["currency"] + dim_currency = df_currency.loc[:, ["currency_id", "currency_code"]] + mappings = { + "GBP": "Pound", + "USD": "US Dollar", + "EUR": "Euro" } -) - -dim_counterparty = df_counterparty.loc[:, "counterparty_id", "counterparty_legal_name", "counterparty_legal_address_line_1", - "counterparty_legal_address_line_2", "counterparty_legal_district", "counterpart_legal_city", - "counterparty_legal_postal_code", "counterparty_legal_country", "counterparty_legal_phone_number"] - -# creates the dim_date dataframe -df_sales = df_sales.loc[:, "agreed_delivery_date"] -df_sales["agreed_delivery_date"] = pd.to_datetime["agreed_delivery_date"] -df_sales["year"] = df_sales["agreed_delivery_date"].dt.year -df_sales["month"] = df_sales["agreed_delivery_date"].dt.month -df_sales["day"] = df_sales["agreed_delivery_date"].dt.day -df_sales["day_of_week"] = df_sales["agreed_delivery_date"].dt.dayofweek -df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() -df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() -df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() + dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) + return dim_currency + + +def create_dim_date(dict_of_df): + df_sales = dict_of_df["sales"] + df_sales = df_sales.loc[:, ["agreed_delivery_date"]] + df_sales["agreed_delivery_date"] = pd.to_datetime["agreed_delivery_date"] + df_sales["year"] = df_sales["agreed_delivery_date"].dt.year + df_sales["month"] = df_sales["agreed_delivery_date"].dt.month + df_sales["day"] = df_sales["agreed_delivery_date"].dt.day + df_sales["day_of_week"] = df_sales["agreed_delivery_date"].dt.dayofweek + df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() + df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() + df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() + dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() + return dim_date # repeat ln 52 - 60 for each column # merge dataframes into one dataframe # remove duplicates -dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() + # TO DO: +# complete dim_date # fact_sales_order -- cgit v1.2.3 From 548b8678e4d5f725e086f0e4eb115c9aa11b55be Mon Sep 17 00:00:00 2001 From: T-Aji Date: Thu, 22 Aug 2024 10:48:54 +0100 Subject: passing tests create_dim_design and create_dim_staff --- src/fact_sales_order.py | 50 ++++++++++++++++++++++++++++++++++++++++++ tests/test_fact_sales_order.py | 40 +++++++++++++++++++++++++++++++++ 2 files changed, 90 insertions(+) create mode 100644 src/fact_sales_order.py create mode 100644 tests/test_fact_sales_order.py diff --git a/src/fact_sales_order.py b/src/fact_sales_order.py new file mode 100644 index 0000000..870a030 --- /dev/null +++ b/src/fact_sales_order.py @@ -0,0 +1,50 @@ +import pandas as pd + + +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 + +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 + +def create_dim_currency(dict_of_df): + df_currency = dict_of_df["currency"] + dim_currency = df_currency.loc[:, ["currency_id", "currency_code"]] + mappings = { + "GBP": "Pound", + "USD": "US Dollar", + "EUR": "Euro" + } + dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) + return dim_currency + + +def create_dim_date(dict_of_df): + df_sales = dict_of_df["sales"] + df_sales = df_sales.loc[:, ["agreed_delivery_date"]] + df_sales["agreed_delivery_date"] = pd.to_datetime["agreed_delivery_date"] + df_sales["year"] = df_sales["agreed_delivery_date"].dt.year + df_sales["month"] = df_sales["agreed_delivery_date"].dt.month + df_sales["day"] = df_sales["agreed_delivery_date"].dt.day + df_sales["day_of_week"] = df_sales["agreed_delivery_date"].dt.dayofweek + df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() + df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() + df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() + dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() + return dim_date + +# repeat ln 52 - 60 for each column +# merge dataframes into one dataframe +# remove duplicates + + + + + +# TO DO: +# complete dim_date +# fact_sales_order diff --git a/tests/test_fact_sales_order.py b/tests/test_fact_sales_order.py new file mode 100644 index 0000000..13196d5 --- /dev/null +++ b/tests/test_fact_sales_order.py @@ -0,0 +1,40 @@ +from src.fact_sales_order import create_dim_design, create_dim_staff +import pandas as pd + +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) + \ No newline at end of file -- cgit v1.2.3 From 21229b09564befcd58363ed7bc1774bbb457ee4b Mon Sep 17 00:00:00 2001 From: T-Aji Date: Thu, 22 Aug 2024 11:03:15 +0100 Subject: passing TestCreateDimCurrency --- tests/test_fact_sales_order.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/tests/test_fact_sales_order.py b/tests/test_fact_sales_order.py index 13196d5..82845d7 100644 --- a/tests/test_fact_sales_order.py +++ b/tests/test_fact_sales_order.py @@ -1,4 +1,4 @@ -from src.fact_sales_order import create_dim_design, create_dim_staff +from src.fact_sales_order import create_dim_design, create_dim_staff, create_dim_currency import pandas as pd class TestCreateDimDesign: @@ -37,4 +37,21 @@ class TestCreateDimStaff: expected_df = pd.DataFrame(data=expected_d) expected_result = expected_df.copy() assert result.equals(expected_result) + +class TestCreateDimCurrency: + def test_dim_currency_returns_dataframe(self): + d = {"currency_id": [1, 2, 3], "currency_code": ["USD", "EUR", "GBP"]} + test_df = {"currency": pd.DataFrame(data=d)} + result = create_dim_currency(test_df) + assert isinstance(result, pd.DataFrame) + + def test_dim_currency_returns_columns_and_values(self): + d = {"currency_id": [1, 2, 3], "currency_code": ["USD", "EUR", "GBP"]} + test_df = {"currency": pd.DataFrame(data=d)} + result = create_dim_currency(test_df) + 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) + expected_result = expected_df.copy() + assert result.equals(expected_result) + \ No newline at end of file -- cgit v1.2.3 From 395731433d9e10eb748fc44669886d8aa80951e1 Mon Sep 17 00:00:00 2001 From: Ang Bel Date: Thu, 22 Aug 2024 11:09:36 +0100 Subject: refactored approach to writing transformation as functions per df. WIP --- src/fact-purchase-table.py | 53 ++++++++++++++++++++++++++-------------------- 1 file changed, 30 insertions(+), 23 deletions(-) diff --git a/src/fact-purchase-table.py b/src/fact-purchase-table.py index 53c0148..91f5077 100644 --- a/src/fact-purchase-table.py +++ b/src/fact-purchase-table.py @@ -6,29 +6,36 @@ import re import pandas as pd -dict_of_df = read_from_s3_subfolder_to_df(tables, extract_bucket(), client=boto3.client("s3")) - - -# iterates through each dataframe in the list of dataframes and assigns them to a variable -df_staff = dict_of_df['staff'] ##no change -df_currency = dict_of_df['currency'] ##scraping API -df_counterparty = dict_of_df['counterparty'] -df_address = dict_of_df['address'] -df_department = dict_of_df['department'] -df_purchase_order = dict_of_df['purchase_order'] +# iterates through each dataframe in the list of dataframes and assigns them to a variable +def get_dfs_from_dict(tables,dictionary=dict_of_df): + for table in tables: + df_staff = dict_of_df['staff'] ##no change + df_currency = dict_of_df['currency'] ##scraping API + df_counterparty = dict_of_df['counterparty'] + df_address = dict_of_df['address'] + df_department = dict_of_df['department'] + df_purchase_order = dict_of_df['purchase_order'] ## dim_staff table is the same across the schemas (no change) -## dim_counterparty table - -## dim_location df_currency --> drops 2 columns -dim_location = df_address.drop(labels=['created_at', 'last_updated'], axis=1).rename(columns={'address_id': 'location_id'}) - -## dim_counterparty -df_prefixed_address = df_address.add_prefix('counterparty_legal_', axis=1) -pd.merge(df_counterparty, - df_prefixed_address, - left_on="legal_address_id", - right_on="address_id", - how="outer") - +## dim_location from address --> drops 2 columns +def create_dim_location(dict_of_df): + dim_location = dict_of_df['address'].drop(labels=['created_at', 'last_updated'], axis=1).rename(columns={'address_id': 'location_id'}) + return dim_location + +## dim_counterparty from address and counterparty +def create_dim_counterparty(dict_of_df): + df_prefixed_address = dict_of_df['address'].add_prefix('counterparty_legal_', axis=1) + pd.merge(dict_of_df['counterparty'], + df_prefixed_address, + left_on="legal_address_id", + right_on="address_id", + how="outer") + +def create_fact_purchase_order(dict_of_df): + df_po = dict_of_df['purchase_order'] + df_po.index.name = 'purchase_record_id' + #df_po['create_date'] = df_po['create_at'].date() + #df_po['create_time'] = df_po['create_at'].time() + df_po['agreed_delivery_date'] = + df_po['agreed_payment_date'] \ No newline at end of file -- cgit v1.2.3 From 8e1893d3943eff65df6517c04b167f7bce0dd200 Mon Sep 17 00:00:00 2001 From: T-Aji Date: Thu, 22 Aug 2024 12:28:13 +0100 Subject: add fact table --- src/fact_sales_order.py | 35 +++++++++++++++++++++++++++++++---- 1 file changed, 31 insertions(+), 4 deletions(-) diff --git a/src/fact_sales_order.py b/src/fact_sales_order.py index 870a030..b657d7d 100644 --- a/src/fact_sales_order.py +++ b/src/fact_sales_order.py @@ -37,14 +37,41 @@ def create_dim_date(dict_of_df): dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() return dim_date -# repeat ln 52 - 60 for each column +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 + df_sales.rename(columns={"staff_id": "sales_staff_id"}) + 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 + +# TO DO: +# complete dim_date from merged fact table # merge dataframes into one dataframe # remove duplicates +# test dim_date and fact_sales_order + -# TO DO: -# complete dim_date -# fact_sales_order -- cgit v1.2.3 From 85c38d9cf43204b1af597fa2762f658e202ac371 Mon Sep 17 00:00:00 2001 From: T-Aji Date: Thu, 22 Aug 2024 12:30:34 +0100 Subject: add fact table --- src/fact-sales-order.py | 50 ------------------------------------------------- 1 file changed, 50 deletions(-) delete mode 100644 src/fact-sales-order.py diff --git a/src/fact-sales-order.py b/src/fact-sales-order.py deleted file mode 100644 index 7921047..0000000 --- a/src/fact-sales-order.py +++ /dev/null @@ -1,50 +0,0 @@ -import pandas as pd - - -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 - -def create_dim_staff(dict_of_df): - staff_department = pd.merge(dict_of_df["staff"], dict_of_df["department"], on='department_id', how="outer") - dim_staff = staff_department.loc[:, ['staff_id', 'first_name', 'last_name', 'department_name', 'location', 'email_address']] - return dim_staff - -def create_dim_currency(dict_of_df): - df_currency = dict_of_df["currency"] - dim_currency = df_currency.loc[:, ["currency_id", "currency_code"]] - mappings = { - "GBP": "Pound", - "USD": "US Dollar", - "EUR": "Euro" - } - dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) - return dim_currency - - -def create_dim_date(dict_of_df): - df_sales = dict_of_df["sales"] - df_sales = df_sales.loc[:, ["agreed_delivery_date"]] - df_sales["agreed_delivery_date"] = pd.to_datetime["agreed_delivery_date"] - df_sales["year"] = df_sales["agreed_delivery_date"].dt.year - df_sales["month"] = df_sales["agreed_delivery_date"].dt.month - df_sales["day"] = df_sales["agreed_delivery_date"].dt.day - df_sales["day_of_week"] = df_sales["agreed_delivery_date"].dt.dayofweek - df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() - df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() - df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() - dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() - return dim_date - -# repeat ln 52 - 60 for each column -# merge dataframes into one dataframe -# remove duplicates - - - - - -# TO DO: -# complete dim_date -# fact_sales_order -- cgit v1.2.3 From c5bc22b0e4e637eb20b1057af937c6eda1def4fa Mon Sep 17 00:00:00 2001 From: Ang Bel Date: Thu, 22 Aug 2024 12:39:03 +0100 Subject: complete code for tables for purchase schema including a scrape for currency table. Test to be done --- src/fact-purchase-table.py | 66 +++++++++++++++++++++++++++++++++------------- 1 file changed, 48 insertions(+), 18 deletions(-) diff --git a/src/fact-purchase-table.py b/src/fact-purchase-table.py index 91f5077..597f104 100644 --- a/src/fact-purchase-table.py +++ b/src/fact-purchase-table.py @@ -4,38 +4,68 @@ import json import boto3 import re import pandas as pd +from datetime import datetime as dt +import requests +from bs4 import BeautifulSoup -# iterates through each dataframe in the list of dataframes and assigns them to a variable -def get_dfs_from_dict(tables,dictionary=dict_of_df): - for table in tables: - df_staff = dict_of_df['staff'] ##no change - df_currency = dict_of_df['currency'] ##scraping API - df_counterparty = dict_of_df['counterparty'] - df_address = dict_of_df['address'] - df_department = dict_of_df['department'] - df_purchase_order = dict_of_df['purchase_order'] - ## dim_staff table is the same across the schemas (no change) ## dim_location from address --> drops 2 columns def create_dim_location(dict_of_df): - dim_location = dict_of_df['address'].drop(labels=['created_at', 'last_updated'], axis=1).rename(columns={'address_id': 'location_id'}) - return dim_location + df_loc = dict_of_df['address'].drop(labels=['created_at', 'last_updated'], axis=1).rename(columns={'address_id': 'location_id'}).set_index('location_id') + return df_loc ## dim_counterparty from address and counterparty def create_dim_counterparty(dict_of_df): df_prefixed_address = dict_of_df['address'].add_prefix('counterparty_legal_', axis=1) - pd.merge(dict_of_df['counterparty'], + df_cp = pd.merge(dict_of_df['counterparty'], df_prefixed_address, left_on="legal_address_id", right_on="address_id", - how="outer") + how="outer").set_index('counterparty_id') + return df_cp +## fact_purchase_order from purchase_order def create_fact_purchase_order(dict_of_df): df_po = dict_of_df['purchase_order'] df_po.index.name = 'purchase_record_id' - #df_po['create_date'] = df_po['create_at'].date() - #df_po['create_time'] = df_po['create_at'].time() - df_po['agreed_delivery_date'] = - df_po['agreed_payment_date'] \ No newline at end of file + 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 + +## dim_date from purchase_order +def create_dim_date(dict_of_df): + sr_date = pd.concat([df['created_date'],df['last_updated_date'],df['agreed_delivery_date'],df['agreed_payment_date']]).sort() + df_date = pd.DataFrame(sr_date,columns='date_id') + 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 + df_date.set_index('date_id') + +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 + +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').set_index('currency_id') + return dim_cur + + + + + -- cgit v1.2.3 From daee22145e8ce27425dd8de941b5ab65e6a619ae Mon Sep 17 00:00:00 2001 From: lian-manonog Date: Thu, 22 Aug 2024 16:03:16 +0100 Subject: Refactored tests for transform lambda - all passing now --- tests/test_transform_lambda.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 5121905..516f83b 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -39,8 +39,8 @@ 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" @@ -56,8 +56,8 @@ 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( @@ -72,3 +72,5 @@ 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) + + -- cgit v1.2.3 From f4bd9e3c85341c0805821728d42d74c19cb16bde Mon Sep 17 00:00:00 2001 From: lian-manonog Date: Thu, 22 Aug 2024 17:06:45 +0100 Subject: wip: wrote pseudocode for lambda handler in writing df to parquet file format and uploading the parquet files --- requirements.txt | 4 ++- src/fact-purchase-table.py | 71 ---------------------------------------------- src/fact_purchase_table.py | 71 ++++++++++++++++++++++++++++++++++++++++++++++ src/transform_lambda.py | 56 +++++++++++++++++++++++++++++++++--- 4 files changed, 126 insertions(+), 76 deletions(-) delete mode 100644 src/fact-purchase-table.py create mode 100644 src/fact_purchase_table.py diff --git a/requirements.txt b/requirements.txt index 62ebbf4..0c81216 100644 --- a/requirements.txt +++ b/requirements.txt @@ -29,4 +29,6 @@ urllib3==2.2.2 Werkzeug==3.0.3 xmltodict==0.13.0 s3fs -pandas \ No newline at end of file +pandas +bs4 +pyarrow \ No newline at end of file diff --git a/src/fact-purchase-table.py b/src/fact-purchase-table.py deleted file mode 100644 index 597f104..0000000 --- a/src/fact-purchase-table.py +++ /dev/null @@ -1,71 +0,0 @@ -from src.transform_lambda import read_from_s3_subfolder_to_df, tables -from src.extract_lambda import extract_bucket -import json -import boto3 -import re -import pandas as pd -from datetime import datetime as dt -import requests -from bs4 import BeautifulSoup - - -## dim_staff table is the same across the schemas (no change) - -## dim_location from address --> drops 2 columns -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'}).set_index('location_id') - return df_loc - -## dim_counterparty from address and counterparty -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="address_id", - how="outer").set_index('counterparty_id') - return df_cp - -## fact_purchase_order from purchase_order -def create_fact_purchase_order(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 - -## dim_date from purchase_order -def create_dim_date(dict_of_df): - sr_date = pd.concat([df['created_date'],df['last_updated_date'],df['agreed_delivery_date'],df['agreed_payment_date']]).sort() - df_date = pd.DataFrame(sr_date,columns='date_id') - 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 - df_date.set_index('date_id') - -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 - -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').set_index('currency_id') - return dim_cur - - - - - diff --git a/src/fact_purchase_table.py b/src/fact_purchase_table.py new file mode 100644 index 0000000..f1d8fe1 --- /dev/null +++ b/src/fact_purchase_table.py @@ -0,0 +1,71 @@ +from bs4 import BeautifulSoup +from src.transform_lambda import read_from_s3_subfolder_to_df, tables +from src.extract_lambda import extract_bucket +import json +import boto3 +import re +import pandas as pd +from datetime import datetime as dt +import requests + + +## dim_staff table is the same across the schemas (no change) + +## dim_location from address --> drops 2 columns +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'}).set_index('location_id') + return df_loc + +## dim_counterparty from address and counterparty +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="address_id", + how="outer").set_index('counterparty_id') + return df_cp + +## fact_purchase_order from purchase_order +def create_fact_purchase_order(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 + +## dim_date from purchase_order +def create_dim_date(dict_of_df): + sr_date = pd.concat([df['created_date'],df['last_updated_date'],df['agreed_delivery_date'],df['agreed_payment_date']]).sort() + df_date = pd.DataFrame(sr_date,columns='date_id') + 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 + df_date.set_index('date_id') + +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 + +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').set_index('currency_id') + return dim_cur + + + + + diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 920a24f..6024a24 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -2,10 +2,11 @@ import json import boto3 import re import pandas as pd - - -def lambda_handler(event, context): - pass +import pyarrow as pa +import pyarrow.parquet as pq +from src.extract_lambda import extract_bucket +from src.fact_purchase_table import * +from src.fact_sales_order import create_dim_staff, create_dim_design, create_fact_sales_order tables = [ @@ -22,6 +23,47 @@ tables = [ "payment_type", ] +def lambda_handler(event, context): + dict_of_df = read_from_s3_subfolder_to_df(tables, extract_bucket(), client=boto3.client("s3")) + common_df_list = [create_dim_counterparty(dict_of_df), + create_dim_date(dict_of_df), + create_dim_location(dict_of_df), + create_dim_currency(dict_of_df), + create_dim_staff(dict_of_df)] + + create_fact_purchase_order() + + f_sales_list = [create_fact_sales_order(), + create_dim_design()] + + + ''' + #dict{ + sales_schema: { + Table_name: df_value, + ...} + payment_schema: + Table_name: df_value, + ...} + purchase_schema: + Table_name: df_value, + ...} + } + + for schema in dict: + for table_name, df_value in schema.items(): + parquet_file = df_value.to_parquet(f'{table_name}.parquet', engine='pyarrow'/'fastparquet'(?)) #we don't know the engine + + s3_key = datetime.strftime( + datetime.today(), f"{schema}/%Y/%m/%d/{table_name}_%H:%M:%S.parquet" + ) + + client.upload_file( + parquet_file, transform_bucket(), s3_key) + ##might need seperate function for easier testing## + ''' + + def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")): table_dfs = {} @@ -34,4 +76,10 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")): table_dfs[table] = pd.concat(list_of_df) return table_dfs +def transform_bucket(client=boto3.client("s3")): + response = client.list_buckets() + bucket_filter = [ + bucket["Name"] for bucket in response["Buckets"] if "transform" in bucket["Name"] + ] + return bucket_filter[0] -- cgit v1.2.3 From a8cadadfe2b96c84a29a252110822ec535a0da7e Mon Sep 17 00:00:00 2001 From: T-Aji Date: Fri, 23 Aug 2024 09:33:17 +0100 Subject: payment schema added --- src/fact_payment.py | 30 ++++++++++++++++++++++++++++++ src/fact_sales_order.py | 18 ++++++++++++++++-- 2 files changed, 46 insertions(+), 2 deletions(-) create mode 100644 src/fact_payment.py diff --git a/src/fact_payment.py b/src/fact_payment.py new file mode 100644 index 0000000..92de67c --- /dev/null +++ b/src/fact_payment.py @@ -0,0 +1,30 @@ +import pandas as pd + +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 + +def create_fact_payment(dict_of_df): + df_payment = dict_of_df["payment"] + df_payment.index.name = "payment_record_id" + df_payment["created_date"] = pd.to_datetime(df_payment["created_at"]).dt.date + df_payment["created_time"] = pd.to_datetime(df_payment["created_at"]).dt.time + df_payment["last_updated_date"] = pd.to_datetime(df_payment["last_updated"]).dt.date + df_payment["last_updated_time"] = pd.to_datetime(df_payment["last_updated"]).dt.time + 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 diff --git a/src/fact_sales_order.py b/src/fact_sales_order.py index b657d7d..425b144 100644 --- a/src/fact_sales_order.py +++ b/src/fact_sales_order.py @@ -44,7 +44,8 @@ def create_fact_sales_order(dict_of_df): 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 - df_sales.rename(columns={"staff_id": "sales_staff_id"}) + pd.merge(dict_of_df["staff"], df_sales["sales_staff_id"], on="staff_id", how="left") + # df_sales.rename(columns={"staff_id": "sales_staff_id"}) fact_sales_order = df_sales.loc[:,[ "sales_record_id", "sales_order_id", @@ -70,7 +71,20 @@ def create_fact_sales_order(dict_of_df): # remove duplicates # test dim_date and fact_sales_order - +def create_sales_star_schema(dict_of_df): + dim_design = create_dim_design(dict_of_df) + dim_staff = create_dim_staff(dict_of_df) + dim_currency = create_dim_currency(dict_of_df) + dim_date = create_dim_date(dict_of_df) + + fact_sales_order = create_fact_sales_order(dict_of_df) + + fact_sales_order = fact_sales_order.merge(dim_design, on='design_id', how='left') + fact_sales_order = fact_sales_order.merge(dim_staff, left_on='sales_staff_id', right_on='staff_id', how='left') + fact_sales_order = fact_sales_order.merge(dim_currency, on='currency_id', how='left') + fact_sales_order = fact_sales_order.merge(dim_date, left_on='agreed_delivery_date', right_on='date_id', how='left') + + return fact_sales_order -- cgit v1.2.3 From 1ba7230de96092e9f401067317d0dfaf881b971b Mon Sep 17 00:00:00 2001 From: T-Aji Date: Fri, 23 Aug 2024 09:55:33 +0100 Subject: dataframes combined into one file --- src/dataframes.py | 238 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 238 insertions(+) create mode 100644 src/dataframes.py diff --git a/src/dataframes.py b/src/dataframes.py new file mode 100644 index 0000000..9ce3be0 --- /dev/null +++ b/src/dataframes.py @@ -0,0 +1,238 @@ +import pandas as pd +from bs4 import BeautifulSoup +from src.transform_lambda import read_from_s3_subfolder_to_df, tables +from src.extract_lambda import extract_bucket +import json +import boto3 +import re +from datetime import datetime as dt +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_dim_transaction(dict_of_df): + pass + +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 + pd.merge(dict_of_df["staff"], df_sales["sales_staff_id"], on="staff_id", how="left") + # df_sales.rename(columns={"staff_id": "sales_staff_id"}) + 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"] = pd.to_datetime(df_payment["created_at"]).dt.date + df_payment["created_time"] = pd.to_datetime(df_payment["created_at"]).dt.time + df_payment["last_updated_date"] = pd.to_datetime(df_payment["last_updated"]).dt.date + df_payment["last_updated_time"] = pd.to_datetime(df_payment["last_updated"]).dt.time + 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 + +## dim_location from address --> drops 2 columns +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'}).set_index('location_id') + return df_loc + +## dim_counterparty from address and counterparty +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="address_id", + how="outer").set_index('counterparty_id') + return df_cp + + +## dim_date from purchase_order +def create_dim_date(dict_of_df): + sr_date = pd.concat([dict_of_df['created_date'],dict_of_df['last_updated_date'],dict_of_df['agreed_delivery_date'],dict_of_df['agreed_payment_date']]).sort() + df_date = pd.DataFrame(sr_date,columns='date_id') + 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 + df_date.set_index('date_id') + +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 + +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').set_index('currency_id') + return dim_cur + + + + + + + +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 + +def create_fact_payment(dict_of_df): + df_payment = dict_of_df["payment"] + df_payment.index.name = "payment_record_id" + df_payment["created_date"] = pd.to_datetime(df_payment["created_at"]).dt.date + df_payment["created_time"] = pd.to_datetime(df_payment["created_at"]).dt.time + df_payment["last_updated_date"] = pd.to_datetime(df_payment["last_updated"]).dt.date + df_payment["last_updated_time"] = pd.to_datetime(df_payment["last_updated"]).dt.time + 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 + +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 + +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 + +def create_dim_currency(dict_of_df): + df_currency = dict_of_df["currency"] + dim_currency = df_currency.loc[:, ["currency_id", "currency_code"]] + mappings = { + "GBP": "Pound", + "USD": "US Dollar", + "EUR": "Euro" + } + dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) + return dim_currency + + +def create_dim_date(dict_of_df): + df_sales = dict_of_df["sales"] + df_sales = df_sales.loc[:, ["agreed_delivery_date"]] + df_sales["agreed_delivery_date"] = pd.to_datetime["agreed_delivery_date"] + df_sales["year"] = df_sales["agreed_delivery_date"].dt.year + df_sales["month"] = df_sales["agreed_delivery_date"].dt.month + df_sales["day"] = df_sales["agreed_delivery_date"].dt.day + df_sales["day_of_week"] = df_sales["agreed_delivery_date"].dt.dayofweek + df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() + df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() + df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() + dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() + return dim_date + + +# TO DO: +# complete dim_date from merged fact table +# merge dataframes into one dataframe +# remove duplicates +# test dim_date and fact_sales_order + +def create_sales_star_schema(dict_of_df): + dim_design = create_dim_design(dict_of_df) + dim_staff = create_dim_staff(dict_of_df) + dim_currency = create_dim_currency(dict_of_df) + dim_date = create_dim_date(dict_of_df) + + fact_sales_order = create_fact_sales_order(dict_of_df) + + fact_sales_order = fact_sales_order.merge(dim_design, on='design_id', how='left') + fact_sales_order = fact_sales_order.merge(dim_staff, left_on='sales_staff_id', right_on='staff_id', how='left') + fact_sales_order = fact_sales_order.merge(dim_currency, on='currency_id', how='left') + fact_sales_order = fact_sales_order.merge(dim_date, left_on='agreed_delivery_date', right_on='date_id', how='left') + + return fact_sales_order + + + +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 + + + + + -- cgit v1.2.3 From 8e20c5c0f43d0f0c4983c8895396de7f62b7c390 Mon Sep 17 00:00:00 2001 From: lian-manonog Date: Fri, 23 Aug 2024 11:06:43 +0100 Subject: Deleted the fact_table schema py files Completed Lambda_handler for transform_lambda - and other helper functions. Testing is still to be done. Need to implement lambda layer to share helper functions across all lambdas --- src/fact_payment.py | 30 ------- src/fact_purchase_table.py | 71 ---------------- src/fact_sales_order.py | 91 --------------------- src/transform_lambda.py | 198 +++++++++++++++++++++++++++++++++++---------- 4 files changed, 157 insertions(+), 233 deletions(-) delete mode 100644 src/fact_payment.py delete mode 100644 src/fact_purchase_table.py delete mode 100644 src/fact_sales_order.py diff --git a/src/fact_payment.py b/src/fact_payment.py deleted file mode 100644 index 92de67c..0000000 --- a/src/fact_payment.py +++ /dev/null @@ -1,30 +0,0 @@ -import pandas as pd - -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 - -def create_fact_payment(dict_of_df): - df_payment = dict_of_df["payment"] - df_payment.index.name = "payment_record_id" - df_payment["created_date"] = pd.to_datetime(df_payment["created_at"]).dt.date - df_payment["created_time"] = pd.to_datetime(df_payment["created_at"]).dt.time - df_payment["last_updated_date"] = pd.to_datetime(df_payment["last_updated"]).dt.date - df_payment["last_updated_time"] = pd.to_datetime(df_payment["last_updated"]).dt.time - 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 diff --git a/src/fact_purchase_table.py b/src/fact_purchase_table.py deleted file mode 100644 index f1d8fe1..0000000 --- a/src/fact_purchase_table.py +++ /dev/null @@ -1,71 +0,0 @@ -from bs4 import BeautifulSoup -from src.transform_lambda import read_from_s3_subfolder_to_df, tables -from src.extract_lambda import extract_bucket -import json -import boto3 -import re -import pandas as pd -from datetime import datetime as dt -import requests - - -## dim_staff table is the same across the schemas (no change) - -## dim_location from address --> drops 2 columns -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'}).set_index('location_id') - return df_loc - -## dim_counterparty from address and counterparty -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="address_id", - how="outer").set_index('counterparty_id') - return df_cp - -## fact_purchase_order from purchase_order -def create_fact_purchase_order(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 - -## dim_date from purchase_order -def create_dim_date(dict_of_df): - sr_date = pd.concat([df['created_date'],df['last_updated_date'],df['agreed_delivery_date'],df['agreed_payment_date']]).sort() - df_date = pd.DataFrame(sr_date,columns='date_id') - 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 - df_date.set_index('date_id') - -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 - -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').set_index('currency_id') - return dim_cur - - - - - diff --git a/src/fact_sales_order.py b/src/fact_sales_order.py deleted file mode 100644 index 425b144..0000000 --- a/src/fact_sales_order.py +++ /dev/null @@ -1,91 +0,0 @@ -import pandas as pd - - -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 - -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 - -def create_dim_currency(dict_of_df): - df_currency = dict_of_df["currency"] - dim_currency = df_currency.loc[:, ["currency_id", "currency_code"]] - mappings = { - "GBP": "Pound", - "USD": "US Dollar", - "EUR": "Euro" - } - dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) - return dim_currency - - -def create_dim_date(dict_of_df): - df_sales = dict_of_df["sales"] - df_sales = df_sales.loc[:, ["agreed_delivery_date"]] - df_sales["agreed_delivery_date"] = pd.to_datetime["agreed_delivery_date"] - df_sales["year"] = df_sales["agreed_delivery_date"].dt.year - df_sales["month"] = df_sales["agreed_delivery_date"].dt.month - df_sales["day"] = df_sales["agreed_delivery_date"].dt.day - df_sales["day_of_week"] = df_sales["agreed_delivery_date"].dt.dayofweek - df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() - df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() - df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() - dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() - return dim_date - -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 - pd.merge(dict_of_df["staff"], df_sales["sales_staff_id"], on="staff_id", how="left") - # df_sales.rename(columns={"staff_id": "sales_staff_id"}) - 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 - -# TO DO: -# complete dim_date from merged fact table -# merge dataframes into one dataframe -# remove duplicates -# test dim_date and fact_sales_order - -def create_sales_star_schema(dict_of_df): - dim_design = create_dim_design(dict_of_df) - dim_staff = create_dim_staff(dict_of_df) - dim_currency = create_dim_currency(dict_of_df) - dim_date = create_dim_date(dict_of_df) - - fact_sales_order = create_fact_sales_order(dict_of_df) - - fact_sales_order = fact_sales_order.merge(dim_design, on='design_id', how='left') - fact_sales_order = fact_sales_order.merge(dim_staff, left_on='sales_staff_id', right_on='staff_id', how='left') - fact_sales_order = fact_sales_order.merge(dim_currency, on='currency_id', how='left') - fact_sales_order = fact_sales_order.merge(dim_date, left_on='agreed_delivery_date', right_on='date_id', how='left') - - return fact_sales_order - - - - diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 6024a24..d30d91d 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -1,13 +1,35 @@ import json import boto3 import re +import logging import pandas as pd import pyarrow as pa import pyarrow.parquet as pq -from src.extract_lambda import extract_bucket -from src.fact_purchase_table import * -from src.fact_sales_order import create_dim_staff, create_dim_design, create_fact_sales_order +from src.dataframes import * +# from src.extract_lambda import extract_bucket, DBConnectionException +import boto3 +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", @@ -24,47 +46,124 @@ tables = [ ] def lambda_handler(event, context): - dict_of_df = read_from_s3_subfolder_to_df(tables, extract_bucket(), client=boto3.client("s3")) - common_df_list = [create_dim_counterparty(dict_of_df), - create_dim_date(dict_of_df), - create_dim_location(dict_of_df), - create_dim_currency(dict_of_df), - create_dim_staff(dict_of_df)] + db = None - create_fact_purchase_order() + try: + db = connect_to_database() + bucket = bucket_name('transform') + existing_s3_files = list_existing_s3_files(bucket) - f_sales_list = [create_fact_sales_order(), - create_dim_design()] - - - ''' - #dict{ - sales_schema: { - Table_name: df_value, - ...} - payment_schema: - Table_name: df_value, - ...} - purchase_schema: - Table_name: df_value, - ...} - } - - for schema in dict: - for table_name, df_value in schema.items(): - parquet_file = df_value.to_parquet(f'{table_name}.parquet', engine='pyarrow'/'fastparquet'(?)) #we don't know the engine - - s3_key = datetime.strftime( - datetime.today(), f"{schema}/%Y/%m/%d/{table_name}_%H:%M:%S.parquet" - ) - - client.upload_file( - parquet_file, transform_bucket(), s3_key) - ##might need seperate function for easier testing## - ''' + dict_of_df = read_from_s3_subfolder_to_df(tables, extract_bucket(), 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: @@ -76,10 +175,27 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")): table_dfs[table] = pd.concat(list_of_df) return table_dfs -def transform_bucket(client=boto3.client("s3")): +def bucket_name(bucket_prefix, client=boto3.client("s3")): response = client.list_buckets() bucket_filter = [ - bucket["Name"] for bucket in response["Buckets"] if "transform" in bucket["Name"] + 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}") + + return existing_files \ No newline at end of file -- cgit v1.2.3 From 2231ea89329bd500f7371b7395f5208f7a86c20e Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Fri, 23 Aug 2024 10:11:40 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 8e20c5c according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/93 --- src/dataframes.py | 293 +++++++++++++++++++++++++---------------- src/transform_lambda.py | 100 +++++++------- tests/test_fact_sales_order.py | 90 ++++++++++--- tests/test_transform_lambda.py | 16 ++- 4 files changed, 319 insertions(+), 180 deletions(-) diff --git a/src/dataframes.py b/src/dataframes.py index 9ce3be0..684f102 100644 --- a/src/dataframes.py +++ b/src/dataframes.py @@ -8,7 +8,7 @@ import re from datetime import datetime as dt import requests -#Table names: +# Table names: # fact_sales_order # fact_purchase_orders # fact_payment @@ -21,9 +21,11 @@ import requests # dim_currency # dim_counterparty + def create_dim_transaction(dict_of_df): pass + def create_fact_sales_order(dict_of_df): df_sales = dict_of_df["sales_order"] df_sales.index.name = "sales_record_id" @@ -33,36 +35,46 @@ def create_fact_sales_order(dict_of_df): df_sales["last_updated_time"] = pd.to_datetime(df_sales["last_updated"]).dt.time pd.merge(dict_of_df["staff"], df_sales["sales_staff_id"], on="staff_id", how="left") # df_sales.rename(columns={"staff_id": "sales_staff_id"}) - 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" - ]] + 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 + +# 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) + 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 @@ -73,69 +85,97 @@ def create_fact_payment(dict_of_df): df_payment["created_time"] = pd.to_datetime(df_payment["created_at"]).dt.time df_payment["last_updated_date"] = pd.to_datetime(df_payment["last_updated"]).dt.date df_payment["last_updated_time"] = pd.to_datetime(df_payment["last_updated"]).dt.time - 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" - ]] + 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 -## dim_location from address --> drops 2 columns + +# dim_location from address --> drops 2 columns + + 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'}).set_index('location_id') + df_loc = ( + dict_of_df["address"] + .drop(labels=["created_at", "last_updated"], axis=1) + .rename(columns={"address_id": "location_id"}) + .set_index("location_id") + ) return df_loc -## dim_counterparty from address and counterparty + +# dim_counterparty from address and counterparty + + 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="address_id", - how="outer").set_index('counterparty_id') + 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="address_id", + how="outer", + ).set_index("counterparty_id") return df_cp -## dim_date from purchase_order +# dim_date from purchase_order def create_dim_date(dict_of_df): - sr_date = pd.concat([dict_of_df['created_date'],dict_of_df['last_updated_date'],dict_of_df['agreed_delivery_date'],dict_of_df['agreed_payment_date']]).sort() - df_date = pd.DataFrame(sr_date,columns='date_id') - 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 - df_date.set_index('date_id') + sr_date = pd.concat( + [ + dict_of_df["created_date"], + dict_of_df["last_updated_date"], + dict_of_df["agreed_delivery_date"], + dict_of_df["agreed_payment_date"], + ] + ).sort() + df_date = pd.DataFrame(sr_date, columns="date_id") + 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 + df_date.set_index("date_id") + 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"})] + 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) + df_cur = sr.str.split(pat=" - ", expand=True).rename( + {0: "currency_code", 1: "currency_name"}, axis=1 + ) return df_cur -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').set_index('currency_id') - return dim_cur - - - - +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" + ).set_index("currency_id") + return dim_cur def create_dim_payment_type(dict_of_df): @@ -143,6 +183,7 @@ def create_dim_payment_type(dict_of_df): dim_payment_type = df_payment_type.loc[:, ["payment_type_id", "payment_type_name"]] return dim_payment_type + def create_fact_payment(dict_of_df): df_payment = dict_of_df["payment"] df_payment.index.name = "payment_record_id" @@ -150,41 +191,57 @@ def create_fact_payment(dict_of_df): df_payment["created_time"] = pd.to_datetime(df_payment["created_at"]).dt.time df_payment["last_updated_date"] = pd.to_datetime(df_payment["last_updated"]).dt.date df_payment["last_updated_time"] = pd.to_datetime(df_payment["last_updated"]).dt.time - 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" - ]] + 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 + 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 + 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']] + 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 + def create_dim_currency(dict_of_df): df_currency = dict_of_df["currency"] dim_currency = df_currency.loc[:, ["currency_id", "currency_code"]] - mappings = { - "GBP": "Pound", - "USD": "US Dollar", - "EUR": "Euro" - } + mappings = {"GBP": "Pound", "USD": "US Dollar", "EUR": "Euro"} dim_currency["currency_name"] = dim_currency["currency_code"].map(mappings) return dim_currency @@ -200,39 +257,49 @@ def create_dim_date(dict_of_df): df_sales["day_name"] = df_sales["agreed_delivery_date"].dt.day_name() df_sales["month_name"] = df_sales["agreed_delivery_date"].dt.month_name() df_sales["quarter"] = df_sales["agreed_delivery_date"].dt.quarter() - dim_date = ["date_id", "year", "month", "day", "day_of_week", "day_name", "month_name", "quarter"] #series.dt.quarter() + dim_date = [ + "date_id", + "year", + "month", + "day", + "day_of_week", + "day_name", + "month_name", + "quarter", + ] # series.dt.quarter() return dim_date -# TO DO: +# TO DO: # complete dim_date from merged fact table # merge dataframes into one dataframe # remove duplicates # test dim_date and fact_sales_order + def create_sales_star_schema(dict_of_df): dim_design = create_dim_design(dict_of_df) dim_staff = create_dim_staff(dict_of_df) dim_currency = create_dim_currency(dict_of_df) dim_date = create_dim_date(dict_of_df) - + fact_sales_order = create_fact_sales_order(dict_of_df) - - fact_sales_order = fact_sales_order.merge(dim_design, on='design_id', how='left') - fact_sales_order = fact_sales_order.merge(dim_staff, left_on='sales_staff_id', right_on='staff_id', how='left') - fact_sales_order = fact_sales_order.merge(dim_currency, on='currency_id', how='left') - fact_sales_order = fact_sales_order.merge(dim_date, left_on='agreed_delivery_date', right_on='date_id', how='left') - - return fact_sales_order + fact_sales_order = fact_sales_order.merge(dim_design, on="design_id", how="left") + fact_sales_order = fact_sales_order.merge( + dim_staff, left_on="sales_staff_id", right_on="staff_id", how="left" + ) + fact_sales_order = fact_sales_order.merge( + dim_currency, on="currency_id", how="left" + ) + fact_sales_order = fact_sales_order.merge( + dim_date, left_on="agreed_delivery_date", right_on="date_id", how="left" + ) + + return fact_sales_order 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 - - - - - diff --git a/src/transform_lambda.py b/src/transform_lambda.py index d30d91d..3e74ee0 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -6,12 +6,14 @@ import pandas as pd import pyarrow as pa import pyarrow.parquet as pq from src.dataframes import * + # from src.extract_lambda import extract_bucket, DBConnectionException import boto3 from botocore.exceptions import ClientError from pg8000.native import Connection, InterfaceError from datetime import datetime + class DBConnectionException(Exception): """Wraps pg8000.native Error or DatabaseError.""" @@ -20,6 +22,7 @@ class DBConnectionException(Exception): self.message = str(e) super().__init__(self.message) + logger = logging.getLogger(__name__) logging.basicConfig( @@ -45,44 +48,45 @@ tables = [ "payment_type", ] + def lambda_handler(event, context): db = None - - try: + + try: db = connect_to_database() - bucket = bucket_name('transform') + bucket = bucket_name("transform") existing_s3_files = list_existing_s3_files(bucket) - dict_of_df = read_from_s3_subfolder_to_df(tables, extract_bucket(), client=boto3.client("s3")) + dict_of_df = read_from_s3_subfolder_to_df( + tables, extract_bucket(), 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)} - + "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)} - + "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 + existing_s3_files, immutable_df_dict, mutable_df_dict, bucket ) - - if not status['uploaded']: + + if not status["uploaded"]: logger.info("No dataframes written to the bucket.") return { - 'statusCode': 204, - "body": json.dumps("No files where uploaded."), + "statusCode": 204, + "body": json.dumps("No files where uploaded."), } - + return { "statusCode": 200, "body": json.dumps( @@ -90,7 +94,7 @@ def lambda_handler(event, context): '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.")} @@ -99,34 +103,38 @@ def lambda_handler(event, context): 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': []} +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) + 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) + 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 + 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) - + status["uploaded"].append(table_name) return status - def retrieve_secrets(): secret_name = "bentley-secrets" region_name = "eu-west-2" @@ -175,19 +183,23 @@ def read_from_s3_subfolder_to_df(tables, bucket, client=boto3.client("s3")): 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"] + 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) + response = client.list_objects_v2(Bucket=bucket_name) if "Contents" in response: existing_files = [obj["Key"] for obj in response["Contents"]] @@ -198,4 +210,4 @@ def list_existing_s3_files(bucket_name, client=boto3.client("s3")): except ClientError as e: logger.error(f"Error listing S3 objects: {e}") - return existing_files \ No newline at end of file + return existing_files diff --git a/tests/test_fact_sales_order.py b/tests/test_fact_sales_order.py index 82845d7..87e3ade 100644 --- a/tests/test_fact_sales_order.py +++ b/tests/test_fact_sales_order.py @@ -1,57 +1,109 @@ -from src.fact_sales_order import create_dim_design, create_dim_staff, create_dim_currency +from src.fact_sales_order import ( + create_dim_design, + create_dim_staff, + create_dim_currency, +) import pandas as pd + 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"]} + 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"]} + 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"]} + 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"]} + 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) + 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"]} + 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_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) + assert result.equals(expected_result) + class TestCreateDimCurrency: def test_dim_currency_returns_dataframe(self): d = {"currency_id": [1, 2, 3], "currency_code": ["USD", "EUR", "GBP"]} test_df = {"currency": pd.DataFrame(data=d)} result = create_dim_currency(test_df) - assert isinstance(result, pd.DataFrame) - + assert isinstance(result, pd.DataFrame) + def test_dim_currency_returns_columns_and_values(self): d = {"currency_id": [1, 2, 3], "currency_code": ["USD", "EUR", "GBP"]} test_df = {"currency": pd.DataFrame(data=d)} result = create_dim_currency(test_df) - expected_d = {"currency_id": [1, 2, 3], "currency_code": ["USD", "EUR", "GBP"], "currency_name": ["US Dollar", "Euro", "Pound"]} + 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) expected_result = expected_df.copy() assert result.equals(expected_result) - - \ No newline at end of file diff --git a/tests/test_transform_lambda.py b/tests/test_transform_lambda.py index 516f83b..a91da92 100644 --- a/tests/test_transform_lambda.py +++ b/tests/test_transform_lambda.py @@ -39,7 +39,12 @@ class TestReadFromS3: ) print(result) expected_df = pd.DataFrame( - np.array([["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"]]), + 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) @@ -56,7 +61,12 @@ class TestReadFromS3: tables, bucket="dummy_buc", client=s3_client ) expected_foods_df = pd.DataFrame( - np.array([["Vegetable", "Sour", "Green", "2022-11-03 14:20:49.962"], ["Berry", "Sweet", "Red", "2022-11-03 14:20:49.962"]]), + 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( @@ -72,5 +82,3 @@ 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) - - -- cgit v1.2.3 From 3ff2182b8256594dfbfe7d8c7480d2ee70067ce5 Mon Sep 17 00:00:00 2001 From: lian-manonog Date: Fri, 23 Aug 2024 11:46:59 +0100 Subject: trying to resolce git index issue conflicts - commiting was the only solution --- src/transform_lambda.py | 13 ++++--------- tests/test_fact_sales_order.py | 4 ++++ 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 3e74ee0..44454e2 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -6,9 +6,6 @@ import pandas as pd import pyarrow as pa import pyarrow.parquet as pq from src.dataframes import * - -# from src.extract_lambda import extract_bucket, DBConnectionException -import boto3 from botocore.exceptions import ClientError from pg8000.native import Connection, InterfaceError from datetime import datetime @@ -34,7 +31,7 @@ logging.basicConfig( logging.getLogger("botocore").setLevel(logging.WARNING) -tables = [ +TABLES = [ "sales_order", "transaction", "payment", @@ -54,12 +51,11 @@ def lambda_handler(event, context): try: db = connect_to_database() - bucket = bucket_name("transform") + bucket = bucket_name('transform') + existing_s3_files = list_existing_s3_files(bucket) - dict_of_df = read_from_s3_subfolder_to_df( - tables, extract_bucket(), client=boto3.client("s3") - ) + 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), @@ -134,7 +130,6 @@ def process_to_parquet_and_upload_to_s3( return status - def retrieve_secrets(): secret_name = "bentley-secrets" region_name = "eu-west-2" diff --git a/tests/test_fact_sales_order.py b/tests/test_fact_sales_order.py index 87e3ade..c4fc9f4 100644 --- a/tests/test_fact_sales_order.py +++ b/tests/test_fact_sales_order.py @@ -1,8 +1,12 @@ +<<<<<<< Updated upstream from src.fact_sales_order import ( create_dim_design, create_dim_staff, create_dim_currency, ) +======= +from fact_sales_order import create_dim_design, create_dim_staff, create_dim_currency +>>>>>>> Stashed changes import pandas as pd -- cgit v1.2.3 From c3e04ab0415ddeedfa1a304296aa0e34fb5f2a1f Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Fri, 23 Aug 2024 10:47:15 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 3ff2182 according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/93 --- src/transform_lambda.py | 9 ++++++--- tests/test_fact_sales_order.py | 16 +++++++++------- 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/src/transform_lambda.py b/src/transform_lambda.py index 44454e2..defa15d 100644 --- a/src/transform_lambda.py +++ b/src/transform_lambda.py @@ -51,11 +51,13 @@ def lambda_handler(event, context): try: db = connect_to_database() - bucket = bucket_name('transform') - + 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")) + 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), @@ -130,6 +132,7 @@ def process_to_parquet_and_upload_to_s3( return status + def retrieve_secrets(): secret_name = "bentley-secrets" region_name = "eu-west-2" diff --git a/tests/test_fact_sales_order.py b/tests/test_fact_sales_order.py index c4fc9f4..dad245e 100644 --- a/tests/test_fact_sales_order.py +++ b/tests/test_fact_sales_order.py @@ -1,13 +1,13 @@ -<<<<<<< Updated upstream +import pandas as pd +from fact_sales_order import create_dim_design, create_dim_staff, create_dim_currency from src.fact_sales_order import ( create_dim_design, create_dim_staff, create_dim_currency, ) -======= -from fact_sales_order import create_dim_design, create_dim_staff, create_dim_currency ->>>>>>> Stashed changes -import pandas as pd +<< << << < Updated upstream +== == == = +>>>>>> > Stashed changes class TestCreateDimDesign: @@ -60,7 +60,8 @@ class TestCreateDimStaff: "email_address": ["Hello", "Bye"], "department_id": ["Hello", "Bye"], } - test_df = {"staff": pd.DataFrame(data=d), "department": pd.DataFrame(data=d2)} + test_df = {"staff": pd.DataFrame( + data=d), "department": pd.DataFrame(data=d2)} result = create_dim_staff(test_df) assert isinstance(result, pd.DataFrame) @@ -77,7 +78,8 @@ class TestCreateDimStaff: "email_address": ["Hello", "Bye"], "department_id": ["Hello", "Bye"], } - test_df = {"staff": pd.DataFrame(data=d), "department": pd.DataFrame(data=d2)} + test_df = {"staff": pd.DataFrame( + data=d), "department": pd.DataFrame(data=d2)} result = create_dim_staff(test_df) expected_d = { "staff_id": ["Hello", "Bye"], -- cgit v1.2.3 From 88f1ef765a9d1113757552ee38ad1bbdb708b629 Mon Sep 17 00:00:00 2001 From: lian-manonog <160282780+lian-manonog@users.noreply.github.com> Date: Fri, 23 Aug 2024 14:53:06 +0100 Subject: Removed redundant empty lines of code --- tests/test_fact_sales_order.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/tests/test_fact_sales_order.py b/tests/test_fact_sales_order.py index dad245e..7592f68 100644 --- a/tests/test_fact_sales_order.py +++ b/tests/test_fact_sales_order.py @@ -5,10 +5,6 @@ from src.fact_sales_order import ( create_dim_staff, create_dim_currency, ) -<< << << < Updated upstream -== == == = ->>>>>> > Stashed changes - class TestCreateDimDesign: def test_dim_design_returns_dataframe(self): -- cgit v1.2.3 From 59035d00133eed3f258f75e3a99ce57cae35989d Mon Sep 17 00:00:00 2001 From: "deepsource-autofix[bot]" <62050782+deepsource-autofix[bot]@users.noreply.github.com> Date: Fri, 23 Aug 2024 13:53:17 +0000 Subject: style: format code with Autopep8, Black and Ruff Formatter This commit fixes the style issues introduced in 88f1ef7 according to the output from Autopep8, Black and Ruff Formatter. Details: https://github.com/ajschofield/de-project-bentley/pull/94 --- tests/test_fact_sales_order.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/tests/test_fact_sales_order.py b/tests/test_fact_sales_order.py index 7592f68..48426b4 100644 --- a/tests/test_fact_sales_order.py +++ b/tests/test_fact_sales_order.py @@ -6,6 +6,7 @@ from src.fact_sales_order import ( create_dim_currency, ) + class TestCreateDimDesign: def test_dim_design_returns_dataframe(self): d = { @@ -56,8 +57,7 @@ class TestCreateDimStaff: "email_address": ["Hello", "Bye"], "department_id": ["Hello", "Bye"], } - test_df = {"staff": pd.DataFrame( - data=d), "department": pd.DataFrame(data=d2)} + test_df = {"staff": pd.DataFrame(data=d), "department": pd.DataFrame(data=d2)} result = create_dim_staff(test_df) assert isinstance(result, pd.DataFrame) @@ -74,8 +74,7 @@ class TestCreateDimStaff: "email_address": ["Hello", "Bye"], "department_id": ["Hello", "Bye"], } - test_df = {"staff": pd.DataFrame( - data=d), "department": pd.DataFrame(data=d2)} + test_df = {"staff": pd.DataFrame(data=d), "department": pd.DataFrame(data=d2)} result = create_dim_staff(test_df) expected_d = { "staff_id": ["Hello", "Bye"], -- cgit v1.2.3