Fix data models with compound fields by moving them to discriminated unions and forcing rebuilds on the models to fix a forward ref to AnyComponent

Author: tipatterson-dev

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "oshconnect"
-version = "0.5.1a9"
+version = "0.5.1a11"
 description = "Library for interfacing with OSH, helping guide visualization efforts, and providing a place to store configurations. Implements OGC CS API Part 3 (Pub/Sub) MQTT topic conventions including :data topics and resource event topics."
 readme = "README.md"
 authors = [

src/oshconnect/__init__.py

@@ -33,7 +33,14 @@
     QuantityRangeSchema,
     TimeRangeSchema,
 )
-from .schema_datamodels import SWEDatastreamRecordSchema, OMJSONDatastreamRecordSchema, JSONCommandSchema
+from .schema_datamodels import (
+    SWEDatastreamRecordSchema,
+    OMJSONDatastreamRecordSchema,
+    SWEJSONCommandSchema,
+    JSONCommandSchema,
+    AnyDatastreamRecordSchema,
+    AnyCommandSchema,
+)
 
 # Event system
 from .events import EventHandler, IEventListener, CallbackListener, DefaultEventTypes, AtomicEventTypes, Event, EventBuilder
@@ -77,7 +84,10 @@
     "TimeRangeSchema",
     "SWEDatastreamRecordSchema",
     "OMJSONDatastreamRecordSchema",
+    "SWEJSONCommandSchema",
     "JSONCommandSchema",
+    "AnyDatastreamRecordSchema",
+    "AnyCommandSchema",
     # Event system
     "EventHandler",
     "IEventListener",

src/oshconnect/resource_datamodels.py

@@ -9,12 +9,12 @@
 import json
 from typing import List, TYPE_CHECKING
 
-from pydantic import BaseModel, ConfigDict, Field, SerializeAsAny, model_validator
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 from shapely import Point
 
 from .api_utils import Link
 from .geometry import Geometry
-from .schema_datamodels import DatastreamRecordSchema, CommandSchema
+from .schema_datamodels import AnyCommandSchema, AnyDatastreamRecordSchema
 from .timemanagement import TimeInstant, TimePeriod
 
 if TYPE_CHECKING:
@@ -227,7 +227,7 @@ class DatastreamResource(BaseModel):
     observed_properties: List[dict] = Field(default_factory=list, alias="observedProperties")
     system_id: str = Field(None, alias="system@id")
     links: List[Link] = Field(None)
-    record_schema: SerializeAsAny[DatastreamRecordSchema] = Field(None, alias="schema")
+    record_schema: AnyDatastreamRecordSchema = Field(None, alias="schema")
 
     @classmethod
     @model_validator(mode="before")
@@ -371,7 +371,7 @@ class ControlStreamResource(BaseModel):
     execution_time: TimePeriod = Field(None, alias="executionTime")
     live: bool = Field(None)
     asynchronous: bool = Field(True, alias="async")
-    command_schema: SerializeAsAny[CommandSchema] = Field(None, alias="schema")
+    command_schema: AnyCommandSchema = Field(None, alias="schema")
     links: List[Link] = Field(None)
 
     def to_csapi_dict(self) -> dict:

src/oshconnect/schema_datamodels.py

@@ -7,13 +7,12 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Union, List, Literal
+from typing import Annotated, Union, List, Literal
 
-from pydantic import BaseModel, Field, SerializeAsAny, field_validator, model_validator, HttpUrl, ConfigDict
+from pydantic import BaseModel, Field, model_validator, HttpUrl, ConfigDict
 
 from .api_utils import Link, URI
-from .csapi4py.constants import ObservationFormat
-from .encoding import Encoding
+from .encoding import JSONEncoding
 from .geometry import Geometry
 from .swe_components import AnyComponent, check_named
 from .timemanagement import TimeInstant
@@ -76,8 +75,16 @@ class SWEJSONCommandSchema(CommandSchema):
     """
     model_config = ConfigDict(populate_by_name=True)
 
-    command_format: str = Field("application/swe+json", alias='commandFormat')
-    encoding: SerializeAsAny[Encoding] = Field(...)
+    # Literal pin powers the discriminated `AnyCommandSchema` union below
+    # and removes the need for a runtime field_validator.
+    command_format: Literal["application/swe+json"] = Field(
+        "application/swe+json", alias='commandFormat')
+    # Concrete subclass instead of `SerializeAsAny[Encoding]` — `JSONEncoding`
+    # is the only Encoding type used in practice, and a concrete type
+    # serializes deterministically without `SerializeAsAny`. If/when more
+    # encoding types arrive, migrate this to a discriminated Union on
+    # `Encoding.type`.
+    encoding: JSONEncoding = Field(...)
     record_schema: AnyComponent = Field(..., alias='recordSchema')
 
     @model_validator(mode="after")
@@ -140,17 +147,17 @@ class DatastreamRecordSchema(BaseModel):
 # docs/osh_spec_deviations.md (swe-json-missing-encoding).
 class SWEDatastreamRecordSchema(DatastreamRecordSchema):
     model_config = ConfigDict(populate_by_name=True)
-    encoding: SerializeAsAny[Encoding] = Field(None)
+    # Multi-Literal acts as the discriminator value(s) for AnyDatastreamRecordSchema
+    # below. Replaces the previous runtime field_validator.
+    obs_format: Literal[
+        "application/swe+json",
+        "application/swe+csv",
+        "application/swe+text",
+        "application/swe+binary",
+    ] = Field(..., alias='obsFormat')
+    encoding: JSONEncoding = Field(None)
     record_schema: AnyComponent = Field(..., alias='recordSchema')
 
-    @field_validator('obs_format')
-    @classmethod
-    def check_check_obs_format(cls, v):
-        if v not in [ObservationFormat.SWE_JSON.value, ObservationFormat.SWE_CSV.value,
-                     ObservationFormat.SWE_TEXT.value, ObservationFormat.SWE_BINARY.value]:
-            raise ValueError('obsFormat must be on of the SWE formats')
-        return v
-
     @model_validator(mode="after")
     def _root_record_schema_requires_name(self):
         check_named(self.record_schema, "SWEDatastreamRecordSchema.recordSchema")
@@ -178,20 +185,15 @@ class OMJSONDatastreamRecordSchema(DatastreamRecordSchema):
     """
     model_config = ConfigDict(populate_by_name=True)
 
-    obs_format: str = Field(ObservationFormat.JSON.value, alias='obsFormat')
+    # Multi-Literal — both wire forms are spec-equivalent for OM+JSON.
+    obs_format: Literal[
+        "application/om+json",
+        "application/json",
+    ] = Field("application/om+json", alias='obsFormat')
     result_schema: AnyComponent = Field(None, alias='resultSchema')
     parameters_schema: AnyComponent = Field(None, alias='parametersSchema')
     result_link: dict = Field(None, alias='resultLink')
 
-    @field_validator('obs_format')
-    @classmethod
-    def _check_obs_format(cls, v):
-        if v not in (ObservationFormat.JSON.value, "application/json"):
-            raise ValueError(
-                f"obsFormat must be 'application/json' or '{ObservationFormat.JSON.value}'"
-            )
-        return v
-
     @model_validator(mode="after")
     def _root_schemas_require_name(self):
         if self.result_schema is not None:
@@ -339,3 +341,30 @@ class SystemHistoryProperties(BaseModel):
     valid_time: list = Field(None)
     parent_system_link: str = Field(None, serialization_alias='parentSystem@link')
     procedure_link: str = Field(None, serialization_alias='procedure@link')
+
+
+# Discriminated unions replace the earlier `SerializeAsAny[<base>]` pattern
+# on resource models. Pydantic dispatches by the literal value of the
+# discriminator field — `obsFormat` / `commandFormat` — so validate and
+# dump round-trip without polymorphism quirks.
+AnyDatastreamRecordSchema = Annotated[
+    Union[SWEDatastreamRecordSchema, OMJSONDatastreamRecordSchema],
+    Field(discriminator='obs_format'),
+]
+"""Public alias for `DatastreamResource.record_schema`. Discriminator: `obs_format`."""
+
+AnyCommandSchema = Annotated[
+    Union[SWEJSONCommandSchema, JSONCommandSchema],
+    Field(discriminator='command_format'),
+]
+"""Public alias for `ControlStreamResource.command_schema`. Discriminator: `command_format`."""
+
+
+# Defense-in-depth: rebuild every container model that forward-references
+# `AnyComponent`. See the matching block in swe_components.py for the
+# `MockValSer` rationale — same fault recurs here because each schema
+# class threads `AnyComponent` through its body.
+SWEJSONCommandSchema.model_rebuild(force=True)
+JSONCommandSchema.model_rebuild(force=True)
+SWEDatastreamRecordSchema.model_rebuild(force=True)
+OMJSONDatastreamRecordSchema.model_rebuild(force=True)

src/oshconnect/streamableresource.py

@@ -1173,28 +1173,14 @@ def add_insert_datastream(self, datastream_schema: DatastreamResource):
             component requires a name.
         :return:
         """
-        print(f'Adding datastream: {datastream_schema.model_dump_json(exclude_none=True, by_alias=True)}')
-        # Make the request to add the datastream
-        # if successful, add the datastream to the system
-        # datastream_record_schema = SWEDatastreamRecordSchema(record_schema=datastream_schema,
-        #                                                      obs_format='application/swe+json', encoding=JSONEncoding())
-        # datastream_resource = DatastreamResource(ds_id="default", name=datastream_schema.name,
-        #                                          output_name=datastream_schema.name,
-        #                                          record_schema=datastream_record_schema,
-        #                                          valid_time=TimePeriod(start=TimeInstant.now_as_time_instant(),
-        #                                                                end=TimeInstant(utc_time=TimeUtils.to_utc_time(
-        #                                                                    "2026-12-31T00:00:00Z"))))
-
         api = self._parent_node.get_api_helper()
-        # print(f'Attempting to create datastream: {datastream_resource.model_dump(by_alias=True, exclude_none=True)}')
         res = api.create_resource(APIResourceTypes.DATASTREAM,
                                   datastream_schema.model_dump_json(by_alias=True, exclude_none=True),
                                   req_headers={'Content-Type': ContentTypes.JSON.value},
                                   parent_res_id=self._resource_id)
 
         if res.ok:
             datastream_id = res.headers['Location'].split('/')[-1]
-            print(f'Resource Location: {datastream_id}')
             datastream_schema.ds_id = datastream_id
         else:
             raise Exception(
@@ -1714,33 +1700,48 @@ def get_status_deque_outbound(self) -> deque:
         return self._outbound_status_deque
 
     def publish_command(self, payload):
-        """Publish ``payload`` to the command MQTT topic. Convenience wrapper for ``publish(payload, 'command')``."""
+        """Publish ``payload`` to the command MQTT topic. Convenience wrapper
+        for ``publish(payload, APIResourceTypes.COMMAND.value)``."""
         self.publish(payload, topic=APIResourceTypes.COMMAND.value)
 
     def publish_status(self, payload):
-        """Publish ``payload`` to the status MQTT topic. Convenience wrapper for ``publish(payload, 'status')``."""
+        """Publish ``payload`` to the status MQTT topic. Convenience wrapper
+        for ``publish(payload, APIResourceTypes.STATUS.value)``."""
         self.publish(payload, topic=APIResourceTypes.STATUS.value)
 
-    def publish(self, payload, topic: str = 'command'):
+    def publish(self, payload, topic: str = APIResourceTypes.COMMAND.value):
         """
         Publishes data to the MQTT topic associated with this control stream resource.
-        :param payload: Data to be published, subclass should determine specifically allowed types
-        :param topic: Specific implementation determines the topic from the provided string
+
+        :param payload: Data to be published; subclass determines specifically allowed types.
+        :param topic: One of ``APIResourceTypes.COMMAND.value`` (``"Command"``,
+            the default) or ``APIResourceTypes.STATUS.value`` (``"Status"``).
+            Pass the enum value rather than a lowercase shorthand — the
+            comparison is case-sensitive against the canonical CS API
+            resource-type strings.
         """
 
         if topic == APIResourceTypes.COMMAND.value:
             self._publish_mqtt(self._topic, payload)
         elif topic == APIResourceTypes.STATUS.value:
             self._publish_mqtt(self._status_topic, payload)
         else:
-            raise ValueError(f"Unsupported topic type {topic} for ControlStream publish().")
+            raise ValueError(
+                f"Unsupported topic {topic!r} for ControlStream publish(); "
+                f"expected {APIResourceTypes.COMMAND.value!r} or "
+                f"{APIResourceTypes.STATUS.value!r}."
+            )
 
     def subscribe(self, topic=None, callback=None, qos=0):
         """
         Subscribes to the MQTT topic associated with this control stream resource.
-        :param topic: Specific implementation determines the topic from the provided string
-        :param callback: Optional callback function to handle incoming messages, if None the default handler is used
-        :param qos: Quality of Service level for the subscription, default is 0
+
+        :param topic: ``None`` (defaults to the command topic),
+            ``APIResourceTypes.COMMAND.value`` (``"Command"``), or
+            ``APIResourceTypes.STATUS.value`` (``"Status"``). Comparison is
+            case-sensitive against the canonical CS API resource-type strings.
+        :param callback: Optional callback function to handle incoming messages, if None the default handler is used.
+        :param qos: Quality of Service level for the subscription, default is 0.
         """
 
         t = None
@@ -1750,7 +1751,11 @@ def subscribe(self, topic=None, callback=None, qos=0):
         elif topic == APIResourceTypes.STATUS.value:
             t = self._status_topic
         else:
-            raise ValueError(f"Invalid topic provided {topic}, must be None or one of 'command' or 'status'.")
+            raise ValueError(
+                f"Invalid topic {topic!r}; must be None, "
+                f"{APIResourceTypes.COMMAND.value!r}, or "
+                f"{APIResourceTypes.STATUS.value!r}."
+            )
 
         if callback is None:
             self._mqtt_client.subscribe(t, qos=qos, msg_callback=self._mqtt_sub_callback)

src/oshconnect/swe_components.py

@@ -11,7 +11,7 @@
 from numbers import Real
 from typing import Union, Any, Literal, Annotated
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator, SerializeAsAny
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from .csapi4py.constants import GeometryTypes
 from .api_utils import UCUMCode, URI
@@ -82,8 +82,7 @@ class VectorSchema(AnyComponentSchema):
     definition: str = Field(...)
     reference_frame: str = Field(..., alias='referenceFrame')
     local_frame: str = Field(None, alias='localFrame')
-    # TODO: VERIFY might need to be moved further down when these are defined
-    coordinates: SerializeAsAny[Union[list[CountSchema], list[QuantitySchema], list[TimeSchema]]] = Field(...)
+    coordinates: Union[list[CountSchema], list[QuantitySchema], list[TimeSchema]] = Field(...)
 
     @model_validator(mode="after")
     def _coordinates_require_name(self):
@@ -273,3 +272,17 @@ class CategoryRangeSchema(AnySimpleComponentSchema):
     ],
     Field(discriminator="type"),
 ]
+
+
+# Rebuild every container model that forward-references AnyComponent.
+# Without this, pydantic leaves a `MockValSer` placeholder on the
+# serializer side — `model_validate` upgrades the validator, but
+# `model_dump`/`model_dump_json` raise
+# `TypeError: 'MockValSer' object is not an instance of 'SchemaSerializer'`.
+# Plain `model_rebuild()` is a no-op (the class reports `model_complete`),
+# so `force=True` is required.
+DataRecordSchema.model_rebuild(force=True)
+VectorSchema.model_rebuild(force=True)
+DataArraySchema.model_rebuild(force=True)
+MatrixSchema.model_rebuild(force=True)
+DataChoiceSchema.model_rebuild(force=True)