get correct epd and work leads for resource forecast reports

epictrack-api/src/api/reports/resource_forecast_report.py

@@ -2,7 +2,7 @@
 
 from calendar import monthrange
 from collections import defaultdict
-from datetime import datetime
+from datetime import datetime, timedelta
 from functools import partial
 from io import BytesIO
 from typing import IO, Dict, List, Tuple
@@ -31,6 +31,7 @@
 from api.models.work import WorkStateEnum
 from api.models.work_type import WorkTypeEnum
 from api.models.role import RoleEnum
+from api.services.staff import StaffService
 from api.services.work_phase import WorkPhaseService
 from api.utils.color_utils import color_with_opacity
 from api.utils.constants import CANADA_TIMEZONE
@@ -703,15 +704,19 @@ def _get_work_team_members(self, work_id) -> Tuple[List[str], str]:
                 StaffWorkRole.role_id.label("role_id"),
             )
         )
+        # Retrieve the responsible EPD and work lead that where active at that end date of the report period
+        # 7 day buffer period is added so if a new EPD/work lead is added at the very end its not their name on report
+        responsible_epd_query = StaffService.find_active_staff_from_special_history(
+            work_id, "responsible_epd_id", self.end_date - timedelta(days=7))
+        work_lead_query = StaffService.find_active_staff_from_special_history(
+            work_id, "work_lead_id", self.end_date - timedelta(days=7))
+        responsible_epd = responsible_epd_query.full_name if responsible_epd_query else ""
+        work_lead = work_lead_query.full_name if work_lead_query else ""
         for work_team_member in work_team_members:
             first_name = work_team_member.first_name
             last_name = work_team_member.last_name
             if work_team_member.role_id == RoleEnum.FN_CAIRT.value:
                 cairt_lead = work_team_member.full_name
-            if work_team_member.role_id == RoleEnum.TEAM_LEAD.value:
-                work_lead = work_team_member.full_name
-            if work_team_member.role_id == RoleEnum.RESPONSIBLE_EPD.value:
-                responsible_epd = work_team_member.full_name
             elif work_team_member.role_id in [RoleEnum.OFFICER_ANALYST.value, RoleEnum.OTHER.value]:
                 staffs.append({"first_name": first_name, "last_name": last_name})
         staffs = sorted(staffs, key=lambda x: x["last_name"])

epictrack-api/src/api/services/staff.py

@@ -17,10 +17,12 @@
 
 import pandas as pd
 from flask import current_app
+from sqlalchemy import Integer, cast, func, or_
 
 from api.exceptions import ResourceExistsError, ResourceNotFoundError, UnprocessableEntityError
 from api.models import Staff, db
 from api.models.position import Position
+from api.models.special_field import EntityEnum, SpecialField
 from api.schemas.response import StaffResponseSchema
 from api.utils.token_info import TokenInfo
 from api.services.keycloak import KeycloakService
@@ -247,3 +249,22 @@ def validate_email_and_get_idir_user_id(cls, email):
         except ValueError:
             current_app.logger.debug(f"Error while reading user details from keycloak with email: {email}")
         return ""
+
+    @classmethod
+    def find_active_staff_from_special_history(cls, work_id, field_name, date):
+        """Returns the staff that was active on the given date by querying the special history table."""
+        query = (
+          db.session.query(Staff)
+          .join(SpecialField, Staff.id == cast(SpecialField.field_value, Integer))
+          .filter(SpecialField.entity == EntityEnum.WORK.value)
+          .filter(SpecialField.entity_id == work_id)
+          .filter(SpecialField.field_name == field_name)
+          .filter(func.date(func.lower(SpecialField.time_range)) <= date)
+          .filter(
+              or_(
+                  func.date(func.upper(SpecialField.time_range)).is_(None),
+                  func.date(func.upper(SpecialField.time_range)) >= date
+              )
+          )
+        )
+        return query.one_or_none()