From 8458279e19c04f851e904699a860882fad5b89dd Mon Sep 17 00:00:00 2001 From: Ross Date: Wed, 17 Dec 2025 17:16:19 +0000 Subject: [PATCH] Enhance constraint handling with safe accessors to avoid Pydantic errors and improve compatibility with legacy data structures --- rota_generator/shifts.py | 206 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 196 insertions(+), 10 deletions(-) diff --git a/rota_generator/shifts.py b/rota_generator/shifts.py index a6392cb..113634e 100644 --- a/rota_generator/shifts.py +++ b/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"" + + +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': , '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: - 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 + # 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)