Upgrade to dev core versions

docs/api_reference/models.md

@@ -0,0 +1,3 @@
+# models
+
+::: oteapi_dlite.models

oteapi_dlite/models.py

@@ -0,0 +1,35 @@
+"""DLite-specific data models."""
+
+from __future__ import annotations
+
+from typing import Annotated, Optional
+
+from oteapi.models import AttrDict
+from pydantic import Field, JsonValue
+
+
+class DLiteResult(AttrDict):
+    """Class for returning values from DLite strategies."""
+
+    collection_id: Annotated[
+        Optional[str], Field(description="A reference to a DLite collection.")
+    ] = None
+
+
+class DLiteConfiguration(DLiteResult):
+    """Data model representing recurring fields necessary in strategy-specific
+    configurations for DLite strategies.
+
+    Note, this data model already includes the `collection_id` field from the
+    `DLiteResult` data model.
+    """
+
+    dlite_settings: Annotated[
+        dict[str, JsonValue],
+        Field(
+            description=(
+                "Settings used by DLite strategies within a single pipeline "
+                "run."
+            )
+        ),
+    ] = {}  # noqa: RUF012

oteapi_dlite/strategies/convert.py

@@ -7,19 +7,16 @@
 
 import importlib
 from collections.abc import Sequence
-from typing import TYPE_CHECKING, Annotated, Optional
+from typing import Annotated, Optional
 
 import dlite
 from oteapi.models import AttrDict, FunctionConfig
 from pydantic import Field
 from pydantic.dataclasses import dataclass
 
-from oteapi_dlite.models import DLiteSessionUpdate
+from oteapi_dlite.models import DLiteResult
 from oteapi_dlite.utils import get_collection, update_collection
 
-if TYPE_CHECKING:  # pragma: no cover
-    from typing import Any
-
 
 class DLiteConvertInputConfig(AttrDict):
     """Configuration for input instance to generic DLite converter.
@@ -64,7 +61,7 @@ class DLiteConvertOutputConfig(AttrDict):
     ] = None
 
 
-class DLiteConvertStrategyConfig(AttrDict):
+class DLiteConvertStrategyConfig(DLiteResult):
     """Configuration for generic DLite converter."""
 
     function_name: Annotated[
@@ -145,35 +142,31 @@ class DLiteConvertStrategy:
 
     """
 
-    convert_config: DLiteConvertConfig
+    function_config: DLiteConvertConfig
 
-    def initialize(
-        self,
-        session: Optional[dict[str, Any]] = None,
-    ) -> DLiteSessionUpdate:
+    def initialize(self) -> DLiteResult:
         """Initialize."""
-        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)
+        return DLiteResult(
+            collection_id=get_collection(
+                self.function_config.configuration.collection_id
+            ).uuid
+        )
 
-    def get(
-        self, session: Optional[dict[str, Any]] = None
-    ) -> DLiteSessionUpdate:
+    def get(self) -> DLiteResult:
         """Execute the strategy.
 
         This method will be called through the strategy-specific endpoint
         of the OTE-API Services.
 
-        Parameters:
-            session: A session-specific dictionary context.
-
         Returns:
             SessionUpdate instance.
         """
-        config = self.convert_config.configuration
+        config = self.function_config.configuration
         module = importlib.import_module(config.module_name, config.package)
         function = getattr(module, config.function_name)
         kwargs = config.kwargs
 
-        coll = get_collection(session)
+        coll = get_collection(config.collection_id)
 
         instances = []
         for i, input_config in enumerate(config.inputs):
@@ -201,4 +194,4 @@ def get(
             coll.add(output_config.label, inst)
 
         update_collection(coll)
-        return DLiteSessionUpdate(collection_id=coll.uuid)
+        return DLiteResult(collection_id=coll.uuid)

oteapi_dlite/strategies/filter.py

@@ -3,21 +3,18 @@
 from __future__ import annotations
 
 import re
-from typing import TYPE_CHECKING, Annotated, Optional
+from typing import Annotated, Optional
 
 from dlite.utils import get_referred_instances
-from oteapi.models import AttrDict, FilterConfig
+from oteapi.models import FilterConfig
 from pydantic import Field
 from pydantic.dataclasses import dataclass
 
-from oteapi_dlite.models import DLiteSessionUpdate
+from oteapi_dlite.models import DLiteResult
 from oteapi_dlite.utils import get_collection, update_collection
 
-if TYPE_CHECKING:  # pragma: no cover
-    from typing import Any
 
-
-class DLiteQueryConfig(AttrDict):
+class DLiteQueryConfig(DLiteResult):
     """Configuration for the DLite filter strategy.
 
     First the `remove_label` and `remove_datamodel` configurations are
@@ -97,22 +94,21 @@ class DLiteFilterStrategy:
 
     **Registers strategies**:
 
-    - `("filterType", "dlite/filter")`
+    - `("filterType", "application/vnd.dlite-filter")`
 
     """
 
     filter_config: DLiteFilterConfig
 
-    def initialize(
-        self,
-        session: Optional[dict[str, Any]] = None,
-    ) -> DLiteSessionUpdate:
+    def initialize(self) -> DLiteResult:
         """Initialize."""
-        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)
+        return DLiteResult(
+            collection_id=get_collection(
+                self.filter_config.configuration.collection_id
+            ).uuid
+        )
 
-    def get(
-        self, session: Optional[dict[str, Any]] = None
-    ) -> DLiteSessionUpdate:
+    def get(self) -> DLiteResult:
         """Execute the strategy."""
         config = self.filter_config.configuration
 
@@ -122,7 +118,7 @@ def get(
         )
 
         instdict = {}  # Map instance labels to [uuid, metaURI]
-        coll = get_collection(session)
+        coll = get_collection(config.collection_id)
         for s, _, o in coll.get_relations(p="_has-uuid"):
             instdict[s] = [o]
         for s, _, o in coll.get_relations(p="_has-meta"):
@@ -170,4 +166,4 @@ def get(
             coll.remove(label)
 
         update_collection(coll)
-        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)
+        return DLiteResult(collection_id=coll.uuid)

oteapi_dlite/strategies/generate.py

@@ -2,17 +2,18 @@
 
 from __future__ import annotations
 
+import json
 import os
 import tempfile
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, Optional
 
 from oteapi.datacache import DataCache
-from oteapi.models import AttrDict, DataCacheConfig, FunctionConfig
+from oteapi.models import DataCacheConfig, FunctionConfig
 from pydantic import Field
 from pydantic.dataclasses import dataclass
 
-from oteapi_dlite.models import DLiteSessionUpdate
+from oteapi_dlite.models import DLiteConfiguration, DLiteResult
 from oteapi_dlite.utils import (
     get_collection,
     get_driver,
@@ -21,10 +22,6 @@
     update_dict,
 )
 
-if TYPE_CHECKING:  # pragma: no cover
-    from typing import Any
-
-
 # Constants
 hasInput = "https://w3id.org/emmo#EMMO_36e69413_8c59_4799_946c_10b05d266e22"
 hasOutput = "https://w3id.org/emmo#EMMO_c4bace1d_4db0_4cd3_87e9_18122bae2840"
@@ -34,7 +31,7 @@
     """Invalid data in knowledge base."""
 
 
-class DLiteStorageConfig(AttrDict):
+class DLiteStorageConfig(DLiteConfiguration):
     """Configuration for a generic DLite storage filter.
 
     The DLite storage driver to can be specified using either the `driver`
@@ -116,12 +113,6 @@
             description="Whether to allow incomplete property mappings.",
         ),
     ] = False
-    collection_id: Annotated[
-        Optional[str],
-        Field(
-            description=("ID of the collection to use."),
-        ),
-    ] = None
     datacache_config: Annotated[
         Optional[DataCacheConfig],
         Field(
@@ -261,42 +252,35 @@
 
     """
 
-    generate_config: DLiteGenerateConfig
+    function_config: DLiteGenerateConfig
 
-    def initialize(
-        self,
-        session: Optional[dict[str, Any]] = None,
-    ) -> DLiteSessionUpdate:
+    def initialize(self) -> DLiteResult:
         """Initialize."""
-        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)
+        return DLiteResult(
+            collection_id=get_collection(
+                self.function_config.configuration.collection_id
+            ).uuid
+        )
 
-    def get(
-        self, session: Optional[dict[str, Any]] = None
-    ) -> DLiteSessionUpdate:
+    def get(self) -> DLiteResult:
         """Execute the strategy.
 
         This method will be called through the strategy-specific endpoint
         of the OTE-API Services.
 
-        Parameters:
-            session: A session-specific dictionary context.
-
         Returns:
             SessionUpdate instance.
         """
-        config = self.generate_config.configuration
+        config = self.function_config.configuration
         cacheconfig = config.datacache_config
 
-        if session is None:
-            session = {}
-
         driver = (
             config.driver
             if config.driver
             else get_driver(mediaType=config.mediaType)
         )
 
-        coll = get_collection(session, config.collection_id)
+        coll = get_collection(config.collection_id)
 
         if config.label:
             inst = coll[config.label]
@@ -338,13 +322,21 @@
             from tripper import RDF
             from tripper.convert import save_container
 
-            # kb_settings = get_settings(session, "tripper.triplestore")
-            # if not kb_settings:
-            #     raise KeyError(
-            #         "The `kb_document_class` configuration requires that a "
-            #         "'tripper.triplestore' settings has been added using the "
-            #         "application/vnd.dlite-settings strategy."
-            #     )
+            kb_settings = config.dlite_settings.get("tripper.triplestore")
+            if isinstance(kb_settings, str):
+                kb_settings = json.loads(kb_settings)
+            if kb_settings and not isinstance(kb_settings, dict):
+                raise ValueError(
+                    "The `kb_document_class` configuration expects a dict "
+                    "with settings for the tripper.triplestore."
+                )
+
+            if TYPE_CHECKING:  # pragma: no cover
+                # This block will only be run by mypy when checking typing
+                assert (
+                    isinstance(kb_settings, dict) or kb_settings is None
+                )  # nosec
+
             # IRI of new individual
             iri = individual_iri(
                 class_iri=config.kb_document_class,
@@ -356,7 +348,10 @@
                 for prop, val in config.kb_document_context.items():
                     triples.append((iri, prop, val))
 
-            ts = get_triplestore(session)
+            ts = get_triplestore(
+                kb_settings=kb_settings,
+                collection_id=config.collection_id,
+            )
             try:
                 if config.kb_document_computation:
                     comput = individual_iri(
@@ -463,7 +458,7 @@
         # other strategies.
 
         update_collection(coll)
-        return DLiteSessionUpdate(collection_id=coll.uuid)
+        return DLiteResult(collection_id=coll.uuid)
 
 
 def individual_iri(