ENH: Add classification and rep_include arguments

src/fmu/dataio/_metadata.py

@@ -97,8 +97,17 @@ def _get_meta_objectdata(
     )
 
 
-def _get_meta_access(access: dict) -> meta.SsdlAccess:
-    return meta.SsdlAccess.model_validate(access)
+def _get_meta_access(dataio: ExportData) -> meta.SsdlAccess:
+    return meta.SsdlAccess(
+        asset=meta.Asset(
+            name=dataio.config.get("access", {}).get("asset", {}).get("name", "")
+        ),
+        classification=dataio._classification,
+        ssdl=meta.Ssdl(
+            access_level=dataio._classification,
+            rep_include=dataio._rep_include,
+        ),
+    )
 
 
 def _get_meta_masterdata(masterdata: dict) -> meta.Masterdata:
@@ -160,7 +169,6 @@ def generate_export_metadata(
     filedata = _get_filedata_provider(dataio, obj, objdata, fmudata, compute_md5)
 
     masterdata = dataio.config.get("masterdata")
-    access = dataio.config.get("access")
 
     metadata = internal.DataClassMeta(
         schema_=TypeAdapter(AnyHttpUrl).validate_strings(SCHEMA),  # type: ignore[call-arg]
@@ -169,7 +177,7 @@ def generate_export_metadata(
         class_=objdata.classname,
         fmu=fmudata.get_metadata() if fmudata else None,
         masterdata=_get_meta_masterdata(masterdata) if masterdata else None,
-        access=_get_meta_access(access) if access else None,
+        access=_get_meta_access(dataio),
         data=_get_meta_objectdata(objdata),
         file=filedata.get_metadata(),
         tracklog=generate_meta_tracklog(),

src/fmu/dataio/dataio.py

@@ -34,7 +34,7 @@
 from .case import InitializeCase
 from .datastructure._internal.internal import AllowedContent
 from .datastructure.configuration import global_configuration
-from .datastructure.meta import meta
+from .datastructure.meta import enums, meta
 from .providers._fmu import FmuProvider, get_fmu_context_from_environment
 
 # DATAIO_EXAMPLES: Final = dataio_examples()
@@ -246,12 +246,17 @@ class ExportData:
         access_ssdl: Optional. A dictionary that will overwrite or append
             to the default ssdl settings read from the config. Example:
             ``{"access_level": "restricted", "rep_include": False}``
+            Deprecated and replaced by 'classification' and 'rep_include' arguments.
 
         casepath: To override the automatic and actual ``rootpath``. Absolute path to
             the case root. If not provided, the rootpath will be attempted parsed from
             the file structure or by other means. See also fmu_context, where "case"
             may need an explicit casepath!
 
+        classification: Optional. Security classification level of the data object.
+            If present it will override the default found in the config.
+            Valid values are either "restricted" or "internal".
+
         config: Required in order to produce valid metadata, either as key (here) or
             through an environment variable. A dictionary with static settings.
             In the standard case this is read from FMU global variables
@@ -317,6 +322,9 @@ class ExportData:
         parent: Optional. This key is required for datatype GridProperty, and
             refers to the name of the grid geometry.
 
+        rep_include: Optional. If True then the data object will be available in REP.
+            Default is False.
+
         runpath: TODO! Optional and deprecated. The relative location of the current run
             root. Optional and will in most cases be auto-detected, assuming that FMU
             folder conventions are followed. For an ERT run e.g.
@@ -400,6 +408,7 @@ class ExportData:
     access_ssdl: dict = field(default_factory=dict)
     aggregation: bool = False
     casepath: Optional[Union[str, Path]] = None
+    classification: Optional[str] = None
     config: dict = field(default_factory=dict)
     content: Optional[Union[dict, str]] = None
     depth_reference: str = "msl"
@@ -414,6 +423,7 @@ class ExportData:
     undef_is_zero: bool = False
     parent: str = ""
     realization: Optional[int] = None  # deprecated
+    rep_include: Optional[bool] = None
     reuse_metadata_rule: Optional[str] = None  # deprecated
     runpath: Optional[Union[str, Path]] = None
     subfolder: str = ""
@@ -436,6 +446,11 @@ class ExportData:
     _fmurun: bool = field(default=False, init=False)
     _reuse_metadata: bool = field(default=False, init=False)
 
+    # Need to store these temporarily in variables until we stop
+    # updating state of the class also on export and generate_metadata
+    _classification: enums.AccessLevel = enums.AccessLevel.internal
+    _rep_include: bool = field(default=False, init=False)
+
     # << NB! storing ACTUAL casepath:
     _rootpath: Path = field(default_factory=Path, init=False)
 
@@ -469,20 +484,97 @@ def __post_init__(self) -> None:
         self._validate_content_key()
         self._validate_and_establish_fmucontext()
         self._validate_workflow_key()
-        self._update_globalconfig_from_settings()
 
         # check state of global config
         self._config_is_valid = global_configuration.is_valid(self.config)
         if self._config_is_valid:
             # TODO: This needs refinement: _config_is_valid should be removed
             self.config = global_configuration.roundtrip(self.config)
 
+        self._classification = self._get_classification()
+        self._rep_include = self._get_rep_include()
+
         self._establish_pwd_rootpath()
         logger.info("Ran __post_init__")
 
+    def _get_classification(self) -> enums.AccessLevel:
+        """
+        Get the security classification as an AccessLevel.
+        The order of how the classification is set is:
+        1. from classification argument if present
+        2. from access_ssdl argument (deprecated) if present
+        3. from access.classification in config (has been mirrored from
+        access.ssdl.access_level if not present)
+
+        """
+        if self.classification is not None:
+            logger.info("Classification is set from input")
+            classification = self.classification
+
+        elif self.access_ssdl and self.access_ssdl.get("access_level"):
+            logger.info("Classification is set from access_ssdl input")
+            classification = self.access_ssdl["access_level"]
+
+        elif self._config_is_valid:
+            logger.info("Classification is set from config")
+            classification = self.config["access"]["classification"]
+        else:
+            # note the one below here will never be used, because that
+            # means the config is invalid and no metadata will be produced
+            logger.info("Using default classification 'internal'")
+            classification = enums.AccessLevel.internal
+
+        if enums.AccessLevel(classification) == enums.AccessLevel.asset:
+            warnings.warn(
+                "The value 'asset' for access.ssdl.access_level is deprecated. "
+                "Please use 'restricted' in input arguments or global variables "
+                "to silence this warning.",
+                FutureWarning,
+            )
+            return enums.AccessLevel.restricted
+        return enums.AccessLevel(classification)
+
+    def _get_rep_include(self) -> bool:
+        """
+        Get the rep_include status.
+        The order of how the staus is set is:
+        1. from rep_include argument if present
+        2. from access_ssdl argument (deprecated) if present
+        3. from access.ssdl.rep_include in config
+        4. default to False if not found
+        """
+        if self.rep_include is not None:
+            logger.debug("rep_include is set from input")
+            return self.rep_include
+
+        if self.access_ssdl and self.access_ssdl.get("rep_include"):
+            logger.debug("rep_include is set from access_ssdl input")
+            return self.access_ssdl["rep_include"]
+
+        if "rep_include" in self.config.get("access", {}).get("ssdl", {}):
+            logger.debug("rep_include is set from config")
+            return self.config["access"]["ssdl"]["rep_include"]
+
+        logger.debug("Using default 'rep_include'=False")
+        return False
+
     def _show_deprecations_or_notimplemented(self) -> None:
         """Warn on deprecated keys or on stuff not implemented yet."""
 
+        if self.access_ssdl:
+            warn(
+                "The 'access_ssdl' argument is deprecated and will be removed in the "
+                "future. Use the more explicit 'classification' and 'rep_include' "
+                "arguments instead.",
+                FutureWarning,
+            )
+            if self.classification is not None or self.rep_include is not None:
+                raise ValueError(
+                    "Using the 'classification' and/or 'rep_include' arguments, "
+                    "in combination with the (legacy) 'access_ssdl' argument "
+                    "is not supported."
+                )
+
         if self.runpath:
             warn(
                 "The 'runpath' key has currently no function. It will be evaluated for "
@@ -643,22 +735,6 @@ def _update_check_settings(self, newsettings: dict) -> None:
         self._validate_workflow_key()
         self._validate_and_establish_fmucontext()
 
-    def _update_globalconfig_from_settings(self) -> None:
-        """A few user settings may update/append the global config directly."""
-        newglobals = deepcopy(self.config)
-
-        if self.access_ssdl:
-            if "ssdl" not in self.config["access"]:
-                newglobals["access"]["ssdl"] = {}
-
-            newglobals["access"]["ssdl"] = deepcopy(self.access_ssdl)
-
-            logger.info(
-                "Updated global config's access.ssdl value: %s", newglobals["access"]
-            )
-
-        self.config = newglobals
-
     def _establish_pwd_rootpath(self) -> None:
         """Establish state variables pwd and the (initial) rootpath.
 
@@ -779,12 +855,9 @@ def generate_metadata(
         logger.info("KW args %s", kwargs)
 
         self._update_check_settings(kwargs)
-        self._update_globalconfig_from_settings()
 
-        self._config_is_valid = global_configuration.is_valid(self.config)
-        if self._config_is_valid:
-            # TODO: This needs refinement: _config_is_valid should be removed
-            self.config = global_configuration.roundtrip(self.config)
+        self._classification = self._get_classification()
+        self._rep_include = self._get_rep_include()
 
         self._check_process_object(obj)  # obj --> self._object
 

src/fmu/dataio/datastructure/configuration/global_configuration.py

@@ -53,25 +53,11 @@ class Ssdl(BaseModel):
     Defines the configuration for the SSDL.
     """
 
-    access_level: enums.AccessLevel = Field(
-        default=enums.AccessLevel.internal,
-    )
-    rep_include: bool = Field(
+    access_level: Optional[enums.AccessLevel] = Field(default=None)
+    rep_include: Optional[bool] = Field(
         default=False,
     )
 
-    @model_validator(mode="after")
-    def _migrate_asset_to_restricted(self) -> Ssdl:
-        if self.access_level == enums.AccessLevel.asset:
-            warnings.warn(
-                "The value 'asset' for access.ssdl.access_level is deprecated. "
-                "Please use 'restricted' in input arguments or global variables "
-                "to silence this warning.",
-                FutureWarning,
-            )
-            self.access_level = enums.AccessLevel.restricted
-        return self
-
 
 class Asset(BaseModel):
     """
@@ -81,21 +67,33 @@ class Asset(BaseModel):
     name: str
 
 
-class Access(BaseModel):
+class Access(BaseModel, use_enum_values=True):
     """
     Manages access configurations, combining asset and SSDL information.
     """
 
     asset: Asset
-    ssdl: Ssdl
+    ssdl: Optional[Ssdl] = Field(default=None)
     classification: Optional[enums.AccessLevel] = Field(default=None)
 
     @model_validator(mode="after")
-    def _classification_mirrors_accesslevel(self) -> Access:
-        # Ideally we want to only copy if the user has NOT
-        # set the classification.
-        # See: https://github.com/equinor/fmu-dataio/issues/540
-        self.classification = self.ssdl.access_level
+    def _validate_classification_ssdl_access_level(self) -> Access:
+        if self.classification and self.ssdl and self.ssdl.access_level:
+            warnings.warn(
+                "The config contains both 'access.ssdl.access_level (deprecated) and "
+                "access.classification. The value from access.classification will be "
+                "used as the default classification. Remove 'access.ssdl.access_level' "
+                "to silence this warning."
+            )
+        if not self.classification:
+            if not (self.ssdl and self.ssdl.access_level):
+                raise ValueError(
+                    "The config doesn't contain any default security classification. "
+                    "Please provide access.classification."
+                )
+            # classification mirrors ssdl.access_level if not present
+            self.classification = self.ssdl.access_level
+
         return self
 
 

tests/conftest.py

@@ -281,10 +281,8 @@ def fixture_globalconfig1():
         ),
         access=global_configuration.Access(
             asset=global_configuration.Asset(name="Test"),
-            ssdl=global_configuration.Ssdl(
-                access_level=global_configuration.enums.AccessLevel.internal,
-                rep_include=False,
-            ),
+            ssdl=global_configuration.Ssdl(rep_include=False),
+            classification=global_configuration.enums.AccessLevel.internal,
         ),
         model=global_configuration.Model(
             name="Test",
@@ -299,7 +297,7 @@ def fixture_globalconfig1():
                 )
             }
         ),
-    ).model_dump()
+    ).model_dump(exclude_none=True)
 
 
 @pytest.fixture(scope="module")