tegwick
388b330809
Share LegacyWorkstreamIdBodyMixin across create schemas; meter POST /tasks/ and /decisions/ workstream_id bodies. State summary uses workplan flow; NextStep dual-writes workplan_* fields alongside legacy workstream_*.
58 lines
No EOL
1.8 KiB
Python
58 lines
No EOL
1.8 KiB
Python
"""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 |