Remove LibHR table management

Why these changes are being introduced:

For reasons beyond the scope of this commit, it has been determined that
having a table like "LibHR Employee Appointments" (LIBHR) distinct from the
more widely used "Employee Appointments" (EA) is more problematic than
it is helpful.  As such, it has been decided that data formerly found
in the LIBHR table will now be directly added and managed in the EA
table.

How this addresses that need:
* Completely removes any tasks associated with the LIBHR table
* Removes any copying of data to the EA table from the LIBHR table

Side effects of this change:
* LIBHR table is no longer a concern for this middleware

Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/IN-1194

Author: ghukill

hrqb/tasks/employee_appointments.py

@@ -5,7 +5,6 @@
 
 from hrqb.base.task import PandasPickleTask, QuickbaseUpsertTask, SQLQueryExtractTask
 from hrqb.utils import md5_hash_from_values, normalize_dataframe_dates
-from hrqb.utils.quickbase import QBClient
 
 
 class ExtractDWEmployeeAppointments(SQLQueryExtractTask):
@@ -18,77 +17,16 @@ def sql_file(self) -> str:
         return "hrqb/tasks/sql/employee_appointments.sql"
 
 
-class ExtractQBLibHREmployeeAppointments(PandasPickleTask):
-    """Query Quickbase for data provided by Library HR about employee appointments."""
-
-    stage = luigi.Parameter("Extract")
-
-    def get_dataframe(self) -> pd.DataFrame:  # pragma: nocover
-        qbclient = QBClient()
-        return qbclient.get_table_as_df(
-            qbclient.get_table_id("LibHR Employee Appointments")
-        )
-
-
-class ExtractQBDepartments(PandasPickleTask):
-    """Query Quickbase for Department data to merge with Library HR data."""
-
-    stage = luigi.Parameter("Extract")
-
-    def get_dataframe(self) -> pd.DataFrame:  # pragma: nocover
-        qbclient = QBClient()
-        return qbclient.get_table_as_df(qbclient.get_table_id("Departments"))
-
-
 class TransformEmployeeAppointments(PandasPickleTask):
     """Combine Data Warehouse and Library HR data for Employee Appointments QB table."""
 
     stage = luigi.Parameter("Transform")
 
     def requires(self) -> list[luigi.Task]:  # pragma: nocover
-        return [
-            ExtractDWEmployeeAppointments(pipeline=self.pipeline),
-            ExtractQBLibHREmployeeAppointments(pipeline=self.pipeline),
-            ExtractQBDepartments(pipeline=self.pipeline),
-        ]
+        return [ExtractDWEmployeeAppointments(pipeline=self.pipeline)]
 
     def get_dataframe(self) -> pd.DataFrame:
-        dw_emp_appts_df = self.named_inputs["ExtractDWEmployeeAppointments"].read()
-        libhr_df = self.named_inputs["ExtractQBLibHREmployeeAppointments"].read()
-        depts_df = self.named_inputs["ExtractQBDepartments"].read()
-
-        # filter libhr data to active appointments, with position IDs
-        libhr_df = libhr_df[(libhr_df["Active"]) & ~(libhr_df["Position ID"].isna())]
-
-        # normalize position id to string and pad zeros
-        libhr_df["Position ID"] = libhr_df["Position ID"].apply(
-            lambda x: str(int(x)).zfill(8)
-        )
-
-        # merge data warehouse data with libhr data to create new employee appointments df
-        emp_appts_df = dw_emp_appts_df.merge(
-            libhr_df[
-                [
-                    "Related Employee MIT ID",
-                    "Position ID",
-                    "Related Supervisor MIT ID",
-                    "HC ID",
-                    "Related Department ID",
-                    "Cost Object",
-                ]
-            ],
-            how="left",
-            left_on=["position_id", "mit_id"],
-            right_on=["Position ID", "Related Employee MIT ID"],
-        )
-
-        # merge on departments to get directorates
-        emp_appts_df = emp_appts_df.merge(
-            depts_df[["Record ID#", "Directorate"]],
-            how="left",
-            left_on="Related Department ID",
-            right_on="Record ID#",
-        )
+        emp_appts_df = self.single_input_dataframe
 
         emp_appts_df = normalize_dataframe_dates(
             emp_appts_df,
@@ -110,20 +48,15 @@ def get_dataframe(self) -> pd.DataFrame:
 
         fields = {
             "mit_id": "MIT ID",
-            "HC ID": "HC ID",
             "employee_type": "Related Employee Type",
             "appt_begin_date": "Begin Date",
             "appt_end_date": "End Date",
-            "Directorate": "Related Directorate",
-            "Related Department ID": "Related Department ID",
-            "Related Supervisor MIT ID": "Supervisor",
             "job_title_long": "Related Job Title",
             "position_title_long": "Related Position Title",
             "job_family": "Job Family",
             "job_subfamily": "Job Subfamily",
             "job_track": "Job Track",
             "position_id": "Position ID",
-            "Cost Object": "Cost Object",
             "exempt": "Exempt / NE",
             "union_name": "Union Name",
             "term_or_perm": "Term or Permanent",

hrqb/tasks/libhr_employee_appointments.py

@@ -24,31 +24,3 @@ def requires(self) -> Iterator[luigi.Task]:  # pragma: no cover
         yield LoadEmployeeLeave(pipeline=self.pipeline_name)
         yield LoadPerformanceReviews(pipeline=self.pipeline_name)
         yield LoadEmployeeLeaveBalances(pipeline=self.pipeline_name)
-
-
-class UpdateLibHRData(HRQBPipelineTask):
-    """Pipeline to load Library HR employee appointment data from static CSV file.
-
-    This pipeline loads the table 'LibHR Employee Appointments', which contains
-    information known only by Library HR, that we cannot get from the data warehouse.
-
-    This Quickbase table is used by the 'Employee Appointments' table to fill in gaps from
-    warehouse data alone.  This pipeline is useful for initial loading and bulk changes,
-    but this table is primarily managed directly in Quickbase by HR staff.
-
-    This pipeline requires a 'csv_filepath' parameter is defined when running, e.g.:
-        pipenv run hrqb --verbose \
-        pipeline -p UpdateLibHRData \
-        --pipeline-parameters=csv_filepath=<PATH/TO/CSV> \
-        run
-    """
-
-    csv_filepath = luigi.Parameter()
-
-    def requires(self) -> Iterator[luigi.Task]:  # pragma: no cover
-        from hrqb.tasks.libhr_employee_appointments import LoadLibHREmployeeAppointments
-
-        yield LoadLibHREmployeeAppointments(
-            pipeline=self.pipeline_name,
-            csv_filepath=self.csv_filepath,
-        )