Merge pull request #185 from MITLibraries/HRQB-51-employee-leave-bala…

…nces

HRQB 51 - New table Employee Leave Balances

hrqb/tasks/employee_leave_balances.py

@@ -0,0 +1,83 @@
+"""hrqb.tasks.employee_leave_balances"""
+
+import luigi  # type: ignore[import-untyped]
+import pandas as pd
+
+from hrqb.base.task import (
+    PandasPickleTask,
+    QuickbaseUpsertTask,
+    SQLQueryExtractTask,
+)
+from hrqb.utils import (
+    md5_hash_from_values,
+    normalize_dataframe_dates,
+)
+
+
+class ExtractDWEmployeeLeaveBalances(SQLQueryExtractTask):
+    """Query Data Warehouse for employee leave balance data."""
+
+    stage = luigi.Parameter("Extract")
+
+    @property
+    def sql_file(self) -> str:
+        return "hrqb/tasks/sql/employee_leave_balances.sql"
+
+
+class TransformEmployeeLeaveBalances(PandasPickleTask):
+
+    stage = luigi.Parameter("Transform")
+
+    def requires(self) -> list[luigi.Task]:
+        return [ExtractDWEmployeeLeaveBalances(pipeline=self.pipeline)]
+
+    def get_dataframe(self) -> pd.DataFrame:
+        dw_leave_balances_df = self.single_input_dataframe
+
+        dw_leave_balances_df = normalize_dataframe_dates(
+            dw_leave_balances_df,
+            ["absence_balance_begin_date", "absence_balance_end_date"],
+        )
+
+        # mint a unique, deterministic value for the merge "Key" field
+        dw_leave_balances_df["key"] = dw_leave_balances_df.apply(
+            lambda row: md5_hash_from_values(
+                [
+                    row.mit_id,
+                    row.balance_type,
+                ]
+            ),
+            axis=1,
+        )
+
+        fields = {
+            "key": "Key",
+            "mit_id": "MIT ID",
+            "balance_type": "Balance Type",
+            "beginning_balance_hours": "Beginning Balance Hours",
+            "deducted_hours": "Deducted Hours",
+            "ending_balance_hours": "Ending Balance Hours",
+            "beginning_balance_days": "Beginning Balance Days",
+            "deducted_days": "Deducted Days",
+            "ending_balance_days": "Ending Balance Days",
+            "absence_balance_begin_date": "Absence Balance Begin Date",
+            "absence_balance_end_date": "Absence Balance End Date",
+        }
+        return dw_leave_balances_df[fields.keys()].rename(columns=fields)
+
+
+class LoadEmployeeLeaveBalances(QuickbaseUpsertTask):
+
+    stage = luigi.Parameter("Load")
+    table_name = "Employee Leave Balances"
+
+    def requires(self) -> list[luigi.Task]:  # pragma: nocover
+        return [TransformEmployeeLeaveBalances(pipeline=self.pipeline)]
+
+    @property
+    def merge_field(self) -> str | None:
+        return "Key"
+
+    @property
+    def input_task_to_load(self) -> str:
+        return "TransformEmployeeLeaveBalances"

hrqb/tasks/pipelines.py

@@ -13,6 +13,7 @@ class FullUpdate(HRQBPipelineTask):
     def requires(self) -> Iterator[luigi.Task]:  # pragma: no cover
         from hrqb.tasks.employee_appointments import LoadEmployeeAppointments
         from hrqb.tasks.employee_leave import LoadEmployeeLeave
+        from hrqb.tasks.employee_leave_balances import LoadEmployeeLeaveBalances
         from hrqb.tasks.employee_salary_history import LoadEmployeeSalaryHistory
         from hrqb.tasks.employees import LoadEmployees
         from hrqb.tasks.performance_reviews import LoadPerformanceReviews
@@ -22,6 +23,7 @@ def requires(self) -> Iterator[luigi.Task]:  # pragma: no cover
         yield LoadEmployeeSalaryHistory(pipeline=self.pipeline_name)
         yield LoadEmployeeLeave(pipeline=self.pipeline_name)
         yield LoadPerformanceReviews(pipeline=self.pipeline_name)
+        yield LoadEmployeeLeaveBalances(pipeline=self.pipeline_name)
 
 
 class UpdateLibHRData(HRQBPipelineTask):

hrqb/tasks/sql/employee_leave_balances.sql

@@ -0,0 +1,28 @@
+/*
+Query for Employee Leave Balances.
+
+CHANGELOG
+    - 2024-09-26 Query created and added
+    - 2024-09-26 Filter to employees with appointment end dates >= 2019
+*/
+
+select
+    b.MIT_ID,
+    bt.ABSENCE_BALANCE_TYPE as BALANCE_TYPE,
+    b.BEGINNING_BALANCE_HOURS,
+    b.DEDUCTED_HOURS,
+    b.ENDING_BALANCE_HOURS,
+    b.BEGINNING_BALANCE_DAYS,
+    b.DEDUCTED_DAYS,
+    b.ENDING_BALANCE_DAYS,
+    b.ABSENCE_BALANCE_BEGIN_DATE,
+    b.ABSENCE_BALANCE_END_DATE
+from HR_ABSENCE_BALANCE b
+inner join HR_ABSENCE_BALANCE_TYPE bt on b.HR_ABSENCE_BALANCE_TYPE_KEY = bt.HR_ABSENCE_BALANCE_TYPE_KEY
+where bt.ABSENCE_BALANCE_TYPE != 'N/A'
+and b.MIT_ID in (
+    select
+        a.MIT_ID
+    from HR_APPOINTMENT_DETAIL a
+    where a.APPT_END_DATE >= TO_DATE('2019-01-01', 'YYYY-MM-DD')
+)

tests/conftest.py

@@ -998,3 +998,50 @@ def single_input_dataframe(self) -> pd.DataFrame:
             )
 
     return UpsertWithDuplicates(pipeline="Checks")
+
+
+@pytest.fixture
+def task_extract_dw_employee_leave_balances_complete(all_tasks_pipeline_name):
+    from hrqb.tasks.employee_leave_balances import ExtractDWEmployeeLeaveBalances
+
+    task = ExtractDWEmployeeLeaveBalances(pipeline=all_tasks_pipeline_name)
+    task.target.write(
+        pd.DataFrame(
+            [
+                {
+                    "mit_id": "123456789",
+                    "balance_type": "MIT Non-Ex Vacation Quota",
+                    "beginning_balance_hours": 80.0,
+                    "deducted_hours": 8.0,
+                    "ending_balance_hours": 72.0,
+                    "beginning_balance_days": 10.0,
+                    "deducted_days": 1.0,
+                    "ending_balance_days": 9.0,
+                    "absence_balance_begin_date": "2006-06-26",
+                    "absence_balance_end_date": "2999-12-31",
+                }
+            ]
+        )
+    )
+    return task
+
+
+@pytest.fixture
+def task_transform_employee_leave_balance_complete(
+    all_tasks_pipeline_name,
+    task_extract_dw_employee_leave_balances_complete,
+):
+    from hrqb.tasks.employee_leave_balances import TransformEmployeeLeaveBalances
+
+    task = TransformEmployeeLeaveBalances(pipeline=all_tasks_pipeline_name)
+    task.run()
+    return task
+
+
+@pytest.fixture
+def task_load_employee_leave_balances(
+    all_tasks_pipeline_name, task_transform_employee_leave_balance_complete
+):
+    from hrqb.tasks.employee_leave_balances import LoadEmployeeLeaveBalances
+
+    return LoadEmployeeLeaveBalances(pipeline=all_tasks_pipeline_name)

tests/tasks/test_employee_leave.py

@@ -42,7 +42,7 @@ def test_task_transform_employee_leave_key_expected_from_row_data(
     )
 
 
-def test_task_load_employee_salary_history_explicit_properties(
+def test_task_load_employee_leave_explicit_properties(
     task_load_employee_leave,
 ):
     assert task_load_employee_leave.merge_field == "Key"

tests/tasks/test_employee_leave_balances.py

@@ -0,0 +1,33 @@
+from hrqb.utils import md5_hash_from_values
+
+
+def test_extract_dw_employee_leave_balances_load_sql_query(
+    task_extract_dw_employee_leave_balances_complete,
+):
+    assert (
+        task_extract_dw_employee_leave_balances_complete.sql_file
+        == "hrqb/tasks/sql/employee_leave_balances.sql"
+    )
+    assert task_extract_dw_employee_leave_balances_complete.sql_query is not None
+
+
+def test_task_transform_employee_leave_balances_key_expected_from_row_data(
+    task_transform_employee_leave_balance_complete,
+):
+    row = task_transform_employee_leave_balance_complete.get_dataframe().iloc[0]
+    assert row["Key"] == md5_hash_from_values(
+        [
+            row["MIT ID"],
+            row["Balance Type"],
+        ]
+    )
+
+
+def test_task_load_employee_leave_balances_explicit_properties(
+    task_load_employee_leave_balances,
+):
+    assert task_load_employee_leave_balances.merge_field == "Key"
+    assert (
+        task_load_employee_leave_balances.input_task_to_load
+        == "TransformEmployeeLeaveBalances"
+    )