"""Shared Pydantic field helpers for workplan / workstream compatibility.""" from __future__ import annotations import uuid from typing import Any from pydantic import AliasChoices, Field, computed_field, model_validator def workplan_id_field(*, default: uuid.UUID | None = None) -> uuid.UUID | None: return Field( default=default, validation_alias=AliasChoices("workplan_id", "workstream_id"), ) class LegacyWorkstreamIdBodyMixin: """Detect POST/PUT bodies that used legacy ``workstream_id`` without ``workplan_id``.""" used_legacy_workstream_id: bool = Field(default=False, exclude=True) @model_validator(mode="before") @classmethod def _detect_legacy_workstream_id(cls, data: Any) -> Any: if isinstance(data, dict): if data.get("workstream_id") is not None and data.get("workplan_id") is None: return {**data, "used_legacy_workstream_id": True} return data class WorkplanIdCompatMixin: """Accept ``workplan_id`` or legacy ``workstream_id`` on input; emit both on output.""" workplan_id: uuid.UUID = workplan_id_field() @computed_field # type: ignore[prop-decorator] @property def workstream_id(self) -> uuid.UUID: return self.workplan_id class WorkplanIdCreateMixin(LegacyWorkstreamIdBodyMixin): workplan_id: uuid.UUID | None = workplan_id_field(default=None) @model_validator(mode="after") def _require_workplan_id(self): if self.workplan_id is None: raise ValueError("workplan_id is required") return self class OptionalWorkplanIdCompatMixin: workplan_id: uuid.UUID | None = workplan_id_field(default=None) @computed_field # type: ignore[prop-decorator] @property def workstream_id(self) -> uuid.UUID | None: return self.workplan_id