Separate out entity models in SOFT and DLite

entities_service/main.py

@@ -10,7 +10,7 @@
 from fastapi import FastAPI, HTTPException, Path, status
 
 from entities_service import __version__
-from entities_service.models import VersionedSOFTEntity
+from entities_service.models import Entity
 from entities_service.service.backend import get_backend
 from entities_service.service.config import CONFIG
 from entities_service.service.logger import setup_logger
@@ -70,7 +70,7 @@ async def lifespan(_: FastAPI):
 
 @APP.get(
     "/{version}/{name}",
-    response_model=VersionedSOFTEntity,
+    response_model=Entity,
     response_model_by_alias=True,
     response_model_exclude_unset=True,
 )

entities_service/models/__init__.py

@@ -6,34 +6,37 @@
 
 from pydantic import ValidationError
 
-from .soft5 import URI_REGEX, SOFT5Entity
+from .dlite_soft5 import DLiteSOFT5Entity
+from .dlite_soft7 import DLiteSOFT7Entity
+from .soft import URI_REGEX
+from .soft5 import SOFT5Entity
 from .soft7 import SOFT7Entity
 
 if TYPE_CHECKING:  # pragma: no cover
     from typing import Literal
 
-VersionedSOFTEntity = SOFT7Entity | SOFT5Entity
-SOFTModelTypes = (SOFT7Entity, SOFT5Entity)
+Entity = SOFT7Entity | SOFT5Entity | DLiteSOFT7Entity | DLiteSOFT5Entity
+EntityType = get_args(Entity)
 
 
 @overload
 def soft_entity(
     *, return_errors: Literal[False] = False, error_msg: str | None = None, **fields
-) -> VersionedSOFTEntity:  # pragma: no cover
+) -> Entity:  # pragma: no cover
     ...
 
 
 @overload
 def soft_entity(
     *, return_errors: Literal[True], error_msg: str | None = None, **fields
-) -> VersionedSOFTEntity | list[ValidationError]:  # pragma: no cover
+) -> Entity | list[ValidationError]:  # pragma: no cover
     ...
 
 
 def soft_entity(*, return_errors: bool = False, error_msg: str | None = None, **fields):
     """Return the correct version of the SOFT Entity."""
     errors = []
-    for versioned_entity_cls in get_args(VersionedSOFTEntity):
+    for versioned_entity_cls in get_args(Entity):
         try:
             new_object = versioned_entity_cls(**fields)
             break
@@ -53,15 +56,15 @@ def soft_entity(*, return_errors: bool = False, error_msg: str | None = None, **
     return new_object  # type: ignore[return-value]
 
 
-def get_uri(entity: VersionedSOFTEntity) -> str:
+def get_uri(entity: Entity) -> str:
     """Return the URI of the entity."""
     if entity.uri is not None:
         return str(entity.uri)
 
     return f"{entity.namespace}/{entity.version}/{entity.name}"
 
 
-def get_version(entity: VersionedSOFTEntity) -> str:
+def get_version(entity: Entity) -> str:
     """Return the version of the entity."""
     if entity.version is not None:
         return entity.version
@@ -72,7 +75,7 @@ def get_version(entity: VersionedSOFTEntity) -> str:
     raise ValueError("Cannot parse URI to get version.")
 
 
-def get_updated_version(entity: VersionedSOFTEntity) -> str:
+def get_updated_version(entity: Entity) -> str:
     """Return the updated version of the entity."""
     current_version = get_version(entity)
 

entities_service/models/dlite_soft.py

@@ -0,0 +1,38 @@
+"""DLite model."""
+
+from __future__ import annotations
+
+from typing import Annotated, Literal
+
+from pydantic import AliasChoices, BaseModel, Field
+from pydantic.networks import AnyHttpUrl
+
+
+class DLiteProperty(BaseModel):
+    """The defining metadata for a DLite Entity's property."""
+
+    ref: Annotated[
+        AnyHttpUrl | None,
+        Field(
+            validation_alias=AliasChoices("$ref", "ref"),
+            serialization_alias="$ref",
+            description=(
+                "Formally a part of type. `$ref` is used together with the `ref` type, "
+                "which is a special datatype for referring to other instances."
+            ),
+        ),
+    ] = None
+
+
+class DLiteEntity(BaseModel):
+    """A DLite Entity."""
+
+    meta: Annotated[
+        Literal["http://onto-ns.com/meta/0.3/EntitySchema"],
+        Field(
+            description=(
+                "URI for the metadata entity. For all entities at onto-ns.com, the "
+                "EntitySchema v0.3 is used."
+            ),
+        ),
+    ] = "http://onto-ns.com/meta/0.3/EntitySchema"

entities_service/models/dlite_soft5.py

@@ -0,0 +1,22 @@
+"""DLite Entity model based on the SOFT5 Entity model."""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from pydantic import Field
+
+from entities_service.models.dlite_soft import DLiteEntity, DLiteProperty
+from entities_service.models.soft5 import SOFT5Entity, SOFT5Property
+
+
+class DLiteSOFT5Property(SOFT5Property, DLiteProperty):
+    """The defining metadata for a (SOFT5-based) DLite Entity's property."""
+
+
+class DLiteSOFT5Entity(SOFT5Entity, DLiteEntity):
+    """A (SOFT5-based) DLite Entity."""
+
+    properties: Annotated[  # type: ignore[assignment]
+        list[DLiteSOFT5Property], Field(description="A list of properties.")
+    ]

entities_service/models/dlite_soft7.py

@@ -0,0 +1,28 @@
+"""DLite Entity model based on the SOFT7 Entity model."""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from pydantic import Field
+
+from entities_service.models.dlite_soft import DLiteEntity, DLiteProperty
+from entities_service.models.soft7 import SOFT7Entity, SOFT7Property
+
+
+class DLiteSOFT7Property(SOFT7Property, DLiteProperty):
+    """The defining metadata for a (SOFT7-based) DLite Entity's property."""
+
+
+class DLiteSOFT7Entity(SOFT7Entity, DLiteEntity):
+    """A (SOFT7-based) DLite Entity."""
+
+    properties: Annotated[  # type: ignore[assignment]
+        dict[str, DLiteSOFT7Property],
+        Field(
+            description=(
+                "A dictionary of properties, mapping the property name to a dictionary "
+                "of metadata defining the property."
+            ),
+        ),
+    ]

entities_service/models/soft.py

@@ -0,0 +1,149 @@
+"""Base (minimum set) models for SOFT entities."""
+
+from __future__ import annotations
+
+import difflib
+import re
+from typing import Annotated, Any
+
+from pydantic import (
+    AliasChoices,
+    BaseModel,
+    ConfigDict,
+    Field,
+    field_validator,
+    model_validator,
+)
+from pydantic.networks import AnyHttpUrl
+
+from entities_service.service.config import CONFIG
+
+URI_REGEX = re.compile(
+    r"^(?P<namespace>https?://.+)/(?P<version>\d(?:\.\d+){0,2})/(?P<name>[^/#?]+)$"
+)
+"""Regular expression to parse a SOFT entity URI."""
+
+
+class SOFTProperty(BaseModel):
+    """The minimum set of defining metadata for a SOFT Entity's property."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    type_: Annotated[
+        str,
+        Field(
+            alias="type",
+            description="The type of the described property, e.g., an integer.",
+        ),
+    ]
+    shape: Annotated[
+        list[str] | None,
+        Field(
+            description=(
+                "The dimension of multi-dimensional properties. This is a list of "
+                "dimension expressions referring to the dimensions defined above. For "
+                "instance, if an entity have dimensions with names `H`, `K`, and `L` "
+                "and a property with shape `['K', 'H+1']`, the property of an instance "
+                "of this entity with dimension values `H=2`, `K=2`, `L=6` will have "
+                "shape `[2, 3]`."
+            ),
+            validation_alias=AliasChoices("dims", "shape"),
+        ),
+    ] = None
+    unit: Annotated[str | None, Field(description="The unit of the property.")] = None
+    description: Annotated[
+        str, Field(description="A human-readable description of the property.")
+    ]
+
+
+class SOFTEntity(BaseModel):
+    """A minimum Field set SOFT Entity to be used in a versioned SOFT Entity model."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    name: Annotated[str | None, Field(description="The name of the entity.")] = None
+    version: Annotated[str | None, Field(description="The version of the entity.")] = (
+        None
+    )
+    namespace: Annotated[
+        AnyHttpUrl | None, Field(description="The namespace of the entity.")
+    ] = None
+    uri: Annotated[
+        AnyHttpUrl | None,
+        Field(
+            description=(
+                "The universal identifier for the entity. This MUST start with the base"
+                " URL."
+            ),
+        ),
+    ] = None
+    description: Annotated[str, Field(description="Description of the entity.")] = ""
+
+    @field_validator("uri", "namespace", mode="after")
+    @classmethod
+    def _validate_base_url(cls, value: AnyHttpUrl) -> AnyHttpUrl:
+        """Validate `uri` and `namespace` starts with the current base URL for the
+        service."""
+        if not str(value).startswith(str(CONFIG.base_url)):
+            error_message = (
+                "This service only works with SOFT entities at " f"{CONFIG.base_url}.\n"
+            )
+            raise ValueError(error_message)
+        return value
+
+    @field_validator("uri", mode="after")
+    @classmethod
+    def _validate_uri(cls, value: AnyHttpUrl) -> AnyHttpUrl:
+        """Validate `uri` is consistent with `name`, `version`, and `namespace`."""
+        if URI_REGEX.match(str(value)) is None:
+            error_message = (
+                "The 'uri' is not a valid SOFT7 entity URI. It must be of the form "
+                f"{str(CONFIG.base_url).rstrip('/')}/{{version}}/{{name}}.\n"
+            )
+            raise ValueError(error_message)
+        return value
+
+    @model_validator(mode="before")
+    @classmethod
+    def _check_cross_dependent_fields(cls, data: Any) -> Any:
+        """Check that `name`, `version`, and `namespace` are all set or all unset."""
+        if (
+            isinstance(data, dict)
+            and any(data.get(_) is None for _ in ("name", "version", "namespace"))
+            and not all(data.get(_) is None for _ in ("name", "version", "namespace"))
+        ):
+            error_message = (
+                "Either all of `name`, `version`, and `namespace` must be set "
+                "or all must be unset.\n"
+            )
+            raise ValueError(error_message)
+
+        if (
+            isinstance(data, dict)
+            and any(data.get(_) is None for _ in ("name", "version", "namespace"))
+            and data.get("uri") is None
+        ):
+            error_message = (
+                "Either `name`, `version`, and `namespace` or `uri` must be set.\n"
+            )
+            raise ValueError(error_message)
+
+        if (
+            isinstance(data, dict)
+            and all(data.get(_) is not None for _ in ("name", "version", "namespace"))
+            and data.get("uri") is not None
+            and data["uri"] != f"{data['namespace']}/{data['version']}/{data['name']}"
+        ):
+            # Ensure that `uri` is consistent with `name`, `version`, and `namespace`.
+            diff = "\n  ".join(
+                difflib.ndiff(
+                    [data["uri"]],
+                    [f"{data['namespace']}/{data['version']}/{data['name']}"],
+                )
+            )
+            error_message = (
+                "The `uri` is not consistent with `name`, `version`, and "
+                f"`namespace`:\n\n  {diff}\n\n"
+            )
+            raise ValueError(error_message)
+        return data