Enhance constraint handling with safe accessors to avoid Pydantic errors and improve compatibility with legacy data structures

rota_generator/shifts.py

@@ -38,6 +38,131 @@ from loguru import logger
 
 logger.add("rota.log", rotation="1 MB", level="DEBUG")
 
+
+def _safe_constraint_info(obj):
+    """Return a tuple (type_name, safe_dump) for logging without calling repr()
+
+    Avoids triggering pydantic __repr__ which can call __getattr__ and raise
+    when models come from mixed pydantic versions. Prefer model_dump() when
+    available; fall back to __dict__ or the class name. Any failure returns a
+    short error string.
+    """
+    tname = type(obj).__name__
+    try:
+        # Only call model_dump on real pydantic BaseModel instances. Some
+        # objects may expose a model_dump attribute or similar but aren't
+        # proper BaseModel instances in this runtime; calling model_dump on
+        # those can raise PydanticUserError. Use isinstance check to be safe.
+        if isinstance(obj, BaseModel) and hasattr(obj, "model_dump"):
+            try:
+                return tname, obj.model_dump()
+            except Exception:
+                # fall through to other strategies
+                pass
+        if hasattr(obj, "__dict__"):
+            return tname, dict(obj.__dict__)
+        return tname, str(type(obj))
+    except Exception as e:
+        return tname, f"<dump-failed: {e!r}>"
+
+
+def _safe_as_dict(obj):
+    """Return a plain dict view of obj without triggering pydantic __getattr__.
+
+    Prefer model_dump(), then __dict__, then if obj is a mapping return it.
+    Returns None if we couldn't obtain a dict safely.
+    """
+    try:
+        # Only call model_dump on actual pydantic BaseModel instances. This
+        # avoids raising PydanticUserError for objects that merely expose a
+        # model_dump attribute but aren't proper BaseModel instances.
+        if isinstance(obj, BaseModel) and hasattr(obj, "model_dump"):
+            try:
+                return obj.model_dump()
+            except Exception:
+                pass
+
+        if isinstance(obj, dict):
+            return obj
+
+        if hasattr(obj, "__dict__"):
+            try:
+                return dict(obj.__dict__)
+            except Exception:
+                pass
+
+    except Exception:
+        # Defensive: anything going wrong, return None so caller can fall back
+        return None
+    return None
+
+
+def _ensure_constraint_weeks(constraint, rota_builder):
+    """Ensure constraint has a `weeks` value, computing from start/end if needed.
+
+    This avoids directly reading attributes on pydantic models which may trigger
+    problematic __getattr__ behavior in mixed-version environments. If we can
+    extract a dict view, use that; otherwise fall back to guarded getattr/setattr.
+    """
+    # Don't mutate unknown constraint objects (some can be pydantic objects
+    # from different versions that break __setattr__). Instead compute the
+    # weeks and cache them on the rota_builder to be used by callers.
+    try:
+        if not hasattr(rota_builder, "_constraint_weeks_cache"):
+            rota_builder._constraint_weeks_cache = {}
+
+        key = id(constraint)
+        if key in rota_builder._constraint_weeks_cache:
+            return
+
+        cdict = _safe_as_dict(constraint)
+        if cdict is not None:
+            weeks = cdict.get("weeks", None)
+            if weeks is None:
+                start = cdict.get("start_date", None)
+                end = cdict.get("end_date", None)
+                computed = rota_builder.get_weeks_by_date_range(start, end)
+                rota_builder._constraint_weeks_cache[key] = computed
+            else:
+                rota_builder._constraint_weeks_cache[key] = weeks
+            return
+
+        # Fallback: try getattr but don't attempt to setattr on the object
+        try:
+            weeks = getattr(constraint, "weeks")
+            rota_builder._constraint_weeks_cache[key] = weeks
+            return
+        except Exception:
+            start = getattr(constraint, "start_date", None)
+            end = getattr(constraint, "end_date", None)
+            computed = rota_builder.get_weeks_by_date_range(start, end)
+            rota_builder._constraint_weeks_cache[key] = computed
+            return
+    except Exception:
+        tname, dump = _safe_constraint_info(constraint)
+        logger.exception("Error accessing/setting 'weeks' on constraint type=%s dump=%s", tname, dump)
+        raise
+
+
+def _get_constraint_weeks(constraint, rota_builder):
+    """Return the weeks list for a constraint using the cache or safe accessors.
+
+    Never calls repr()/__repr__ on the constraint.
+    """
+    if hasattr(rota_builder, "_constraint_weeks_cache"):
+        wk = rota_builder._constraint_weeks_cache.get(id(constraint), None)
+        if wk is not None:
+            return wk
+
+    cdict = _safe_as_dict(constraint)
+    if cdict is not None:
+        return cdict.get("weeks", None)
+
+    try:
+        return getattr(constraint, "weeks")
+    except Exception:
+        return None
+
 ShiftName = str
 DayStr = str
 WeekInt = int
@@ -94,9 +219,16 @@ class BaseShiftConstraint(BaseModel):
         return weeks
 
 
-#class ShiftConstraint(BaseShiftConstraint):
-#    name: str
-#    options: Any = None
+class ShiftConstraint(BaseShiftConstraint):
+    """Compatibility wrapper for legacy constraint dicts of the form
+    {'name': <constraint_name>, 'options': {...}}.
+
+    The DB historically stored constraints as name/options pairs. When we
+    encounter that shape we convert into this wrapper so validation passes
+    and the builder code can still inspect `name`/`options` if needed.
+    """
+    name: str
+    options: Any = None
 
 class NightConstraint(BaseShiftConstraint):
     options: Any = None
@@ -237,7 +369,12 @@ class SingleShift(BaseModel):
     hard_constrain_shift: bool = True
     bank_holidays_only: bool = False
     #constraint: list[ShiftConstraint] = []
-    constraints: List[BaseModel] = Field(default_factory=list)
+    # NOTE: avoid typing this as List[BaseModel] — instructing pydantic to
+    # validate items as `BaseModel` can cause it to attempt to instantiate
+    # `BaseModel` directly which raises a PydanticUserError. Use the
+    # concrete BaseShiftConstraint base type so pydantic will validate into
+    # actual constraint subclasses instead of BaseModel itself.
+    constraints: List["BaseShiftConstraint"] = Field(default_factory=list)
 
     start_date: datetime.date | None = None
     end_date: datetime.date | None = None
@@ -251,6 +388,42 @@ class SingleShift(BaseModel):
     )
 
     def __init__(self, **data: Any):
+        # Backwards-compat: accept legacy `constraints` list entries in the
+        # form {'name': ..., 'options': {...}} and convert them to concrete
+        # pydantic constraint models before validation. This avoids errors
+        # where the DB stored legacy name/options pairs.
+        if "constraints" in data and isinstance(data["constraints"], (list, tuple)):
+            converted = []
+            _name_map = {
+                "pre": PreShiftConstraint,
+                "post": PostShiftConstraint,
+                "night": NightConstraint,
+                "balance_across_groups": BalanceAcrossGroupsConstraint,
+                "min_summed_grade_by_shifts_per_day": MinSummedGradeByShiftsPerDayConstraint,
+                "limit_grade_number": LimitGradeNumberConstraint,
+                "minimum_grade_number": MinimumGradeNumberConstraint,
+                "require_remote_site_presence": RequireRemoteSitePresenceConstraint,
+                "max_shifts_per_week": MaxShiftsPerWeekConstraint,
+                "max_shifts_per_week_block": MaxShiftsPerWeekBlockConstraint,
+            }
+            for item in data["constraints"]:
+                if isinstance(item, dict) and "name" in item:
+                    cname = item.get("name")
+                    opts = item.get("options") or {}
+                    cls = _name_map.get(cname)
+                    if cls is not None:
+                        try:
+                            inst = cls.model_validate(opts) if hasattr(cls, "model_validate") else cls(**(opts or {}))
+                        except Exception:
+                            # fallback to generic wrapper
+                            inst = ShiftConstraint(name=cname, options=opts)
+                    else:
+                        inst = ShiftConstraint(name=cname, options=opts)
+                    converted.append(inst)
+                else:
+                    converted.append(item)
+            data["constraints"] = converted
+
         if "constraint" in data:
             raise ValueError(
                 "The 'constraint' field on SingleShift has been deprecated and removed. "
@@ -1287,8 +1460,19 @@ class RotaBuilder(object):
 
 
         for constraint in self.constraint_options_model.min_summed_grade_by_shifts_per_day:
+            # Safely extract weeks/days without invoking pydantic __getattr__
+            cdict = _safe_as_dict(constraint)
+            if cdict is not None:
+                weeks = cdict.get("weeks") if cdict.get("weeks") is not None else self.weeks
+                days_list = cdict.get("days") if cdict.get("days") is not None else self.days
+            else:
+                try:
                     weeks = constraint.weeks if constraint.weeks is not None else self.weeks
                     days_list = constraint.days if constraint.days is not None else self.days
+                except Exception:
+                    tname, dump = _safe_constraint_info(constraint)
+                    logger.exception("Error accessing 'weeks' or 'days' on constraint type=%s dump=%s", tname, dump)
+                    raise
             for week in weeks:
                 for day in days_list:
                     # Only consider if all shifts are present that day
@@ -3352,7 +3536,8 @@ class RotaBuilder(object):
                         ignore_shifts = constraint.ignore_shifts + [constraint_shift.name]
                     else:
                         ignore_shifts = constraint.ignore_shifts
-                    if constraint.weeks is not None and week not in constraint.weeks:
+                    weeks = _get_constraint_weeks(constraint, self)
+                    if weeks is not None and week not in weeks:
                         continue
                     date_main = self.get_date_by_week_day(week, day)
                     for n in range(0, constraint.days):
@@ -3402,7 +3587,8 @@ class RotaBuilder(object):
                         ignore_shifts = constraint.ignore_shifts + [constraint_shift.name]
                     else:
                         ignore_shifts = constraint.ignore_shifts
-                    if constraint.weeks is not None and week not in constraint.weeks:
+                    weeks = _get_constraint_weeks(constraint, self)
+                    if weeks is not None and week not in weeks:
                         continue
                     date_main = self.get_date_by_week_day(week, day)
                     for n in range(0, constraint.days):
@@ -4149,8 +4335,8 @@ class RotaBuilder(object):
                 self.sites.add(site)
 
             for constraint in s.constraints:
-                if constraint.weeks is None:
-                    constraint.weeks = self.get_weeks_by_date_range(constraint.start_date, constraint.end_date)
+                # Use a safe accessor to avoid triggering pydantic __getattr__ bugs
+                _ensure_constraint_weeks(constraint, self)
 
         self.shift_counts = defaultdict(int)
         self.shift_worker_counts = defaultdict(int)