Add employee appointment integrity checks

hrqb/tasks/employee_leave.py

@@ -7,6 +7,7 @@
 import pandas as pd
 
 from hrqb.base.task import (
+    HRQBTask,
     PandasPickleTask,
     QuickbaseUpsertTask,
     SQLQueryExtractTask,
@@ -104,6 +105,18 @@ def get_dataframe(self) -> pd.DataFrame:
         }
         return leaves_df[fields.keys()].rename(columns=fields)
 
+    @HRQBTask.integrity_check
+    def all_rows_have_employee_appointments(self, output_df: pd.DataFrame) -> None:
+        missing_appointment_count = len(
+            output_df[output_df["Related Employee Appointment"].isna()]
+        )
+        if missing_appointment_count > 0:
+            message = (
+                f"{missing_appointment_count} rows are missing an Employee "
+                f"Appointment for task '{self.name}'"
+            )
+            raise ValueError(message)
+
 
 class LoadEmployeeLeave(QuickbaseUpsertTask):
 

hrqb/tasks/employee_salary_history.py

@@ -3,7 +3,12 @@
 import luigi  # type: ignore[import-untyped]
 import pandas as pd
 
-from hrqb.base.task import PandasPickleTask, QuickbaseUpsertTask, SQLQueryExtractTask
+from hrqb.base.task import (
+    HRQBTask,
+    PandasPickleTask,
+    QuickbaseUpsertTask,
+    SQLQueryExtractTask,
+)
 from hrqb.tasks.employee_appointments import TransformEmployeeAppointments
 from hrqb.utils import md5_hash_from_values, normalize_dataframe_dates
 
@@ -146,6 +151,18 @@ def _set_base_salary_change_percent(self, salary_df: pd.DataFrame) -> pd.DataFra
         )
         return new_salary_df
 
+    @HRQBTask.integrity_check
+    def all_rows_have_employee_appointments(self, output_df: pd.DataFrame) -> None:
+        missing_appointment_count = len(
+            output_df[output_df["Related Employee Appointment"].isna()]
+        )
+        if missing_appointment_count > 0:
+            message = (
+                f"{missing_appointment_count} rows are missing an Employee "
+                f"Appointment for task '{self.name}'"
+            )
+            raise ValueError(message)
+
 
 class LoadEmployeeSalaryHistory(QuickbaseUpsertTask):
     table_name = luigi.Parameter("Employee Salary History")