ENH: Generalize schema configuration and dumping

schemas/0.8.0/fmu_results.json

@@ -10879,7 +10879,7 @@
       "type": "object"
     }
   },
-  "$id": "https://main-fmu-schemas-prod.radix.equinor.com/schemas/0.8.0/fmu_results.json",
+  "$id": "https://main-fmu-schemas-dev.radix.equinor.com/schemas/0.8.0/fmu_results.json",
   "$schema": "https://json-schema.org/draft/2020-12/schema",
   "discriminator": {
     "propertyName": "class"

src/fmu/dataio/_definitions.py

@@ -2,24 +2,11 @@
 
 from __future__ import annotations
 
+import os
+from abc import ABC, abstractmethod
 from enum import Enum
-from typing import Final
-
-
-class FmuResultsSchema:
-    DEV_ROOT: Final[str] = "https://main-fmu-schemas-dev.radix.equinor.com/schemas"
-    PROD_ROOT: Final[str] = "https://main-fmu-schemas-prod.radix.equinor.com/schemas"
-    VERSION: Final[str] = "0.8.0"
-    FILENAME: Final[str] = "fmu_results.json"
-    DEV_URL: Final[str] = f"{DEV_ROOT}/{VERSION}/{FILENAME}"
-    PROD_URL: Final[str] = f"{PROD_ROOT}/{VERSION}/{FILENAME}"
-
-    @staticmethod
-    def url() -> str:
-        """This method is meant to return the `PROD_URL` or `DEV_URL` under relevant
-        circumstances."""
-        return FmuResultsSchema.PROD_URL
-
+from pathlib import Path
+from typing import Any, Final
 
 SOURCE: Final = "fmu"
 
@@ -32,6 +19,78 @@ class ConfigurationError(ValueError):
     pass
 
 
+class FmuSchemas:
+    """These URLs can be constructed programmatically from radixconfig.yaml if need be:
+
+        {cfg.components[].name}-{cfg.metadata.name}-{spec.environments[].name}
+
+    As they are unlikely to change they are hardcoded here.
+    """
+
+    DEV_URL: Final[str] = "https://main-fmu-schemas-dev.radix.equinor.com"
+    PROD_URL: Final[str] = "https://main-fmu-schemas-prod.radix.equinor.com"
+    PATH: Final[Path] = Path("schemas")
+
+
+class SchemaBase(ABC):
+    VERSION: str
+    """The current version of the schema."""
+
+    FILENAME: str
+    """The filename, i.e. schema.json."""
+
+    PATH: Path
+    """The on-disk _and_ URL path following the domain, i.e:
+
+        schemas/0.1.0/schema.json
+
+    This path should _always_ have `FmuSchemas.PATH` as its first parent.
+    This determines the on-disk and URL location of this schema file. A
+    trivial example is:
+
+        PATH: Path = FmuSchemas.PATH / VERSION / FILENAME
+
+    """
+
+    @classmethod
+    def __init_subclass__(cls, **kwargs: dict[str, Any]) -> None:
+        super().__init_subclass__(**kwargs)
+        for attr in ("VERSION", "FILENAME", "PATH"):
+            if not hasattr(cls, attr):
+                raise TypeError(f"Subclass {cls.__name__} must define '{attr}'")
+
+    @classmethod
+    def url(cls) -> str:
+        """Returns the URL this file will reside at, based upon class variables set here
+        and in FmuSchemas."""
+        DEV_URL = f"{FmuSchemas.DEV_URL}/{cls.PATH}"
+        PROD_URL = f"{FmuSchemas.PROD_URL}/{cls.PATH}"
+
+        if os.environ.get("SCHEMA_RELEASE", None):
+            return PROD_URL
+        return DEV_URL
+
+    @staticmethod
+    @abstractmethod
+    def dump() -> dict[str, Any]:
+        """
+        Dumps the export root model to JSON format for schema validation and
+        usage in FMU data structures.
+
+        To update the schema:
+            1. Run the following CLI command to dump the updated schema:
+                `./tools/update_schema`.
+            2. Check the diff for changes. Adding fields usually indicates non-breaking
+                changes and is generally safe. However, if fields are removed, it could
+                indicate breaking changes that may affect dependent systems. Perform a
+                quality control (QC) check to ensure these changes do not break existing
+                implementations.
+                If changes are satisfactory and do not introduce issues, commit
+                them to maintain schema consistency.
+        """
+        raise NotImplementedError
+
+
 class ValidFormats(Enum):
     surface = {
         "irap_binary": ".gri",

src/fmu/dataio/_metadata.py

@@ -7,9 +7,6 @@
 
 from typing import TYPE_CHECKING, Final
 
-from pydantic import AnyHttpUrl, TypeAdapter
-
-from ._definitions import SOURCE, FmuResultsSchema
 from ._logging import null_logger
 from ._model import fields, schema
 from ._model.global_configuration import GlobalConfiguration
@@ -105,10 +102,7 @@ def generate_export_metadata(
 
     objdata = objectdata_provider_factory(obj, dataio, product)
 
-    return schema.InternalObjectMetadata(
-        schema_=TypeAdapter(AnyHttpUrl).validate_strings(FmuResultsSchema.url()),  # type: ignore[call-arg]
-        version=FmuResultsSchema.VERSION,
-        source=SOURCE,
+    return schema.InternalObjectMetadata(  # type: ignore[call-arg]
         class_=objdata.classname,
         fmu=_get_meta_fmu(fmudata) if fmudata else None,
         masterdata=(

src/fmu/dataio/_model/__init__.py

@@ -1,6 +1,6 @@
-from .root import Root, dump
+from .root import FmuResultsSchema, Root
 
 __all__ = [
-    "dump",
+    "FmuResultsSchema",
     "Root",
 ]

src/fmu/dataio/_model/root.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from pathlib import Path
 from typing import TYPE_CHECKING, Dict, List, Literal, TypeVar, Union
 
 from pydantic import (
@@ -12,7 +13,7 @@
 from pydantic.json_schema import GenerateJsonSchema
 from typing_extensions import Annotated
 
-from fmu.dataio._definitions import FmuResultsSchema
+from fmu.dataio._definitions import FmuSchemas, SchemaBase
 
 from .data import AnyData
 from .enums import FMUClass
@@ -198,103 +199,97 @@ def __get_pydantic_json_schema__(
         return json_schema
 
 
-class FmuResultsJsonSchema(GenerateJsonSchema):
-    contractual: Final[list[str]] = [
-        "access",
-        "class",
-        "data.alias",
-        "data.bbox",
-        "data.content",
-        "data.format",
-        "data.geometry",
-        "data.grid_model",
-        "data.is_observation",
-        "data.is_prediction",
-        "data.name",
-        "data.offset",
-        "data.product.name",
-        "data.seismic.attribute",
-        "data.spec.columns",
-        "data.stratigraphic",
-        "data.stratigraphic_alias",
-        "data.tagname",
-        "data.time",
-        "data.vertical_domain",
-        "file.checksum_md5",
-        "file.relative_path",
-        "file.size_bytes",
-        "fmu.aggregation.operation",
-        "fmu.aggregation.realization_ids",
-        "fmu.case",
-        "fmu.context.stage",
-        "fmu.iteration.name",
-        "fmu.iteration.uuid",
-        "fmu.model",
-        "fmu.realization.id",
-        "fmu.realization.is_reference",
-        "fmu.realization.name",
-        "fmu.realization.uuid",
-        "fmu.workflow",
-        "masterdata",
-        "source",
-        "tracklog.datetime",
-        "tracklog.event",
-        "tracklog.user.id",
-        "version",
-    ]
-
-    def _remove_format_path(self, obj: T) -> T:
-        """
-        Removes entries with key "format" and value "path" from dictionaries. This
-        adjustment is necessary because JSON Schema does not recognize the "format":
-        "path", while OpenAPI does. This function is used in contexts where OpenAPI
-        specifications are not applicable.
-        """
-
-        if isinstance(obj, dict):
-            return {
-                k: self._remove_format_path(v)
-                for k, v in obj.items()
-                if not (k == "format" and v == "path")
-            }
-
-        if isinstance(obj, list):
-            return [self._remove_format_path(element) for element in obj]
-
-        return obj
-
-    def generate(
-        self,
-        schema: Mapping[str, Any],
-        mode: Literal["validation", "serialization"] = "validation",
-    ) -> dict[str, Any]:
-        json_schema = super().generate(schema, mode=mode)
-        json_schema["$schema"] = self.schema_dialect
-        json_schema["$id"] = FmuResultsSchema.url()
-        json_schema["$contractual"] = self.contractual
-
-        # sumo-core's validator does not recognize these.
-        del json_schema["discriminator"]["mapping"]
-        del json_schema["$defs"]["AnyData"]["discriminator"]["mapping"]
-        del json_schema["$defs"]["AnyProduct"]["discriminator"]["mapping"]
-
-        return self._remove_format_path(json_schema)
-
+class FmuResultsSchema(SchemaBase):
+    """The main metadata export describing the results."""
+
+    VERSION: str = "0.8.0"
+    FILENAME: str = "fmu_results.json"
+    PATH: Path = FmuSchemas.PATH / VERSION / FILENAME
+
+    class FmuResultsGenerateJsonSchema(GenerateJsonSchema):
+        contractual: Final[list[str]] = [
+            "access",
+            "class",
+            "data.alias",
+            "data.bbox",
+            "data.content",
+            "data.format",
+            "data.geometry",
+            "data.grid_model",
+            "data.is_observation",
+            "data.is_prediction",
+            "data.name",
+            "data.offset",
+            "data.product.name",
+            "data.seismic.attribute",
+            "data.spec.columns",
+            "data.stratigraphic",
+            "data.stratigraphic_alias",
+            "data.tagname",
+            "data.time",
+            "data.vertical_domain",
+            "file.checksum_md5",
+            "file.relative_path",
+            "file.size_bytes",
+            "fmu.aggregation.operation",
+            "fmu.aggregation.realization_ids",
+            "fmu.case",
+            "fmu.context.stage",
+            "fmu.iteration.name",
+            "fmu.iteration.uuid",
+            "fmu.model",
+            "fmu.realization.id",
+            "fmu.realization.is_reference",
+            "fmu.realization.name",
+            "fmu.realization.uuid",
+            "fmu.workflow",
+            "masterdata",
+            "source",
+            "tracklog.datetime",
+            "tracklog.event",
+            "tracklog.user.id",
+            "version",
+        ]
 
-def dump() -> dict:
-    """
-    Dumps the export root model to JSON format for schema validation and
-    usage in FMU data structures.
-
-    To update the schema:
-        1. Run the following CLI command to dump the updated schema:
-            `./tools/update_schema`
-        2. Check the diff for changes. Adding fields usually indicates non-breaking
-            changes and is generally safe. However, if fields are removed, it could
-            indicate breaking changes that may affect dependent systems. Perform a
-            quality control (QC) check to ensure these changes do not break existing
-            implementations.
-            If changes are satisfactory and do not introduce issues, commit
-            them to maintain schema consistency.
-    """
-    return Root.model_json_schema(schema_generator=FmuResultsJsonSchema)
+        def _remove_format_path(self, obj: T) -> T:
+            """
+            Removes entries with key "format" and value "path" from dictionaries. This
+            adjustment is necessary because JSON Schema does not recognize the "format":
+            "path", while OpenAPI does. This function is used in contexts where OpenAPI
+            specifications are not applicable.
+            """
+
+            if isinstance(obj, dict):
+                return {
+                    k: self._remove_format_path(v)
+                    for k, v in obj.items()
+                    if not (k == "format" and v == "path")
+                }
+
+            if isinstance(obj, list):
+                return [self._remove_format_path(element) for element in obj]
+
+            return obj
+
+        def generate(
+            self,
+            schema: Mapping[str, Any],
+            mode: Literal["validation", "serialization"] = "validation",
+        ) -> dict[str, Any]:
+            json_schema = super().generate(schema, mode=mode)
+            json_schema["$schema"] = self.schema_dialect
+            json_schema["$id"] = FmuResultsSchema.url()
+            json_schema["$contractual"] = self.contractual
+
+            # sumo-core's validator does not recognize these.
+            del json_schema["discriminator"]["mapping"]
+            del json_schema["$defs"]["AnyData"]["discriminator"]["mapping"]
+            del json_schema["$defs"]["AnyProduct"]["discriminator"]["mapping"]
+
+            return self._remove_format_path(json_schema)
+
+    @staticmethod
+    def dump() -> dict[str, Any]:
+        return Root.model_json_schema(
+            schema_generator=FmuResultsSchema.FmuResultsGenerateJsonSchema
+        )