started to add support for deposit step and added useful method for SoftwareMetadata

Author: Michael Fritzsche

src/hermes/commands/__init__.py

@@ -15,5 +15,5 @@
 # from hermes.commands.curate.base import HermesCurateCommand
 from hermes.commands.harvest.base import HermesHarvestCommand
 # from hermes.commands.process.base import HermesProcessCommand
-# from hermes.commands.deposit.base import HermesDepositCommand
+from hermes.commands.deposit.base import HermesDepositCommand
 # from hermes.commands.postprocess.base import HermesPostprocessCommand

src/hermes/commands/cli.py

@@ -16,7 +16,7 @@
 # from hermes.commands import (HermesHelpCommand, HermesVersionCommand, HermesCleanCommand,
 #                              HermesHarvestCommand, HermesProcessCommand, HermesCurateCommand,
 #                              HermesDepositCommand, HermesPostprocessCommand, HermesInitCommand)
-from hermes.commands import HermesHarvestCommand
+from hermes.commands import HermesDepositCommand, HermesHarvestCommand
 from hermes.commands.base import HermesCommand
 
 
@@ -45,7 +45,7 @@ def main() -> None:
             HermesHarvestCommand(parser),
             # HermesProcessCommand(parser),
             # HermesCurateCommand(parser),
-            # HermesDepositCommand(parser),
+            HermesDepositCommand(parser),
             # HermesPostprocessCommand(parser),
     ):
         if command.settings_class is not None:

src/hermes/commands/deposit/base.py

@@ -7,15 +7,13 @@
 
 import abc
 import argparse
-import json
-import sys
 
 from pydantic import BaseModel
 
 from hermes.commands.base import HermesCommand, HermesPlugin
-from hermes.model.context import CodeMetaContext
-from hermes.model.path import ContextPath
-from hermes.model.errors import HermesValidationError
+from hermes.model.context_manager import HermesContext
+from hermes.model import SoftwareMetadata
+from hermes.model.error import HermesValidationError
 
 
 class BaseDepositPlugin(HermesPlugin):
@@ -24,16 +22,19 @@ class BaseDepositPlugin(HermesPlugin):
     TODO: describe workflow... needs refactoring to be less stateful!
     """
 
-    def __init__(self, command, ctx):
-        self.command = command
-        self.ctx = ctx
-
     def __call__(self, command: HermesCommand) -> None:
         """Initiate the deposition process.
 
         This calls a list of additional methods on the class, none of which need to be implemented.
         """
         self.command = command
+        self.ctx = HermesContext()
+
+        self.ctx.prepare_step("curate")
+        self.metadata = SoftwareMetadata.load_from_cache(self.ctx, "result")
+        self.ctx.finalize_step("curate")
+
+        self.ctx.prepare_step("deposit")
 
         self.prepare()
         self.map_metadata()
@@ -106,7 +107,7 @@ def publish(self) -> None:
         pass
 
 
-class _DepositSettings(BaseModel):
+class DepositSettings(BaseModel):
     """Generic deposition settings."""
 
     target: str = ""
@@ -116,7 +117,7 @@ class HermesDepositCommand(HermesCommand):
     """ Deposit the curated metadata to repositories. """
 
     command_name = "deposit"
-    settings_class = _DepositSettings
+    settings_class = DepositSettings
 
     def init_command_parser(self, command_parser: argparse.ArgumentParser) -> None:
         command_parser.add_argument('--file', '-f', nargs=1, action='append',
@@ -128,26 +129,12 @@ def __call__(self, args: argparse.Namespace) -> None:
         self.args = args
         plugin_name = self.settings.target
 
-        ctx = CodeMetaContext()
-        codemeta_file = ctx.get_cache("curate", ctx.hermes_name)
-        if not codemeta_file.exists():
-            self.log.error("You must run the 'curate' command before deposit")
-            sys.exit(1)
-
-        codemeta_path = ContextPath("codemeta")
-        with open(codemeta_file) as codemeta_fh:
-            ctx.update(codemeta_path, json.load(codemeta_fh))
-
         try:
-            plugin_func = self.plugins[plugin_name](self, ctx)
-
+            plugin_func = self.plugins[plugin_name]()
+            plugin_func(self)
         except KeyError as e:
             self.log.error("Plugin '%s' not found.", plugin_name)
             self.errors.append(e)
-
-        try:
-            plugin_func(self)
-
         except HermesValidationError as e:
             self.log.error("Error while executing %s: %s", plugin_name, e)
             self.errors.append(e)

src/hermes/commands/deposit/file.py

@@ -11,22 +11,17 @@
 from pydantic import BaseModel
 
 from hermes.commands.deposit.base import BaseDepositPlugin
-from hermes.model.path import ContextPath
 
 
 class FileDepositSettings(BaseModel):
-    filename: str = 'hermes.json'
+    filename: str = 'codemeta.json'
 
 
 class FileDepositPlugin(BaseDepositPlugin):
     settings_class = FileDepositSettings
 
-    def map_metadata(self) -> None:
-        self.ctx.update(ContextPath.parse('deposit.file'), self.ctx['codemeta'])
-
     def publish(self) -> None:
         file_config = self.command.settings.file
-        output_data = self.ctx['deposit.file']
 
         with open(file_config.filename, 'w') as deposition_file:
-            json.dump(output_data, deposition_file, indent=2)
+            json.dump(self.metadata.compact(), deposition_file, indent=2)

src/hermes/commands/deposit/invenio.py

@@ -17,11 +17,10 @@
 import requests
 from pydantic import BaseModel
 
-from hermes.commands.deposit.base import BaseDepositPlugin, HermesDepositCommand
+from hermes.commands.deposit.base import BaseDepositPlugin
 from hermes.commands.deposit.error import DepositionUnauthorizedError
 from hermes.error import MisconfigurationError
-from hermes.model.context import CodeMetaContext
-from hermes.model.path import ContextPath
+from hermes.model.context_manager import HermesContext
 from hermes.utils import hermes_doi, hermes_user_agent
 
 
@@ -258,11 +257,13 @@ class InvenioDepositPlugin(BaseDepositPlugin):
     invenio_resolver_class = InvenioResolver
     settings_class = InvenioDepositSettings
 
-    def __init__(self, command: HermesDepositCommand, ctx: CodeMetaContext, client=None, resolver=None) -> None:
-        super().__init__(command, ctx)
+    def __init__(self) -> None:
+        super().__init__()
 
-        self.invenio_context_path = ContextPath.parse(f"deposit.{self.platform_name}")
         self.invenio_ctx = None
+
+    def __call__(self, command, *, client=None, resolver=None):
+        self.command = command
         self.config = getattr(self.command.settings, self.platform_name)
 
         if client is None:
@@ -292,7 +293,9 @@ def __init__(self, command: HermesDepositCommand, ctx: CodeMetaContext, client=N
         self.resolver = resolver or self.invenio_resolver_class(self.client)
         self.links = {}
 
-    # TODO: Populate some data structure here? Or move more of this into __init__?
+        super().__call__(command)
+
+    # TODO: Populate some data structure here? Or move more of this into __init__.py?
     def prepare(self) -> None:
         """Prepare the deposition on an Invenio-based platform.
 
@@ -305,49 +308,42 @@ def prepare(self) -> None:
         - check access modalities (access right, access conditions, embargo data, existence
           of license)
         - check whether required configuration options are present
-        - update ``self.ctx`` with metadata collected during the checks
+        - update ``self.metadata`` with metadata collected during the checks
         """
 
         rec_id = self.config.record_id
         doi = self.config.doi
 
-        try:
-            codemeta_identifier = self.ctx["codemeta.identifier"]
-        except KeyError:
-            codemeta_identifier = None
-
+        codemeta_identifier = self.metadata.get("identifier", None)
         rec_id, rec_meta = self.resolver.resolve_latest_id(
             record_id=rec_id, doi=doi, codemeta_identifier=codemeta_identifier
         )
 
-        version = self.ctx["codemeta"].get("version")
+        version = self.metadata["version"]
         if rec_meta and (version == rec_meta.get("version")):
             raise ValueError(f"Version {version} already deposited.")
 
-        self.ctx.update(self.invenio_context_path['latestRecord'], {'id': rec_id, 'metadata': rec_meta})
-
-        license = self._get_license_identifier()
-        self.ctx.update(self.invenio_context_path["license"], license)
-
-        communities = self._get_community_identifiers()
-        self.ctx.update(self.invenio_context_path["communities"], communities)
+        deposition_data = {}
+        deposition_data["latestRecord"] = {'id': rec_id, 'metadata': rec_meta}
+        deposition_data["license"] = self._get_license_identifier()
+        deposition_data["communities"] = self._get_community_identifiers()
 
         access_right, embargo_date, access_conditions = self._get_access_modalities(license)
-        self.ctx.update(self.invenio_context_path["access_right"], access_right)
-        self.ctx.update(self.invenio_context_path["embargo_date"], embargo_date)
-        self.ctx.update(self.invenio_context_path["access_conditions"], access_conditions)
+        deposition_data["access_right"] = access_right
+        deposition_data["embargo_date"] = embargo_date
+        deposition_data["access_conditions"] = access_conditions
 
-        self.invenio_ctx = self.ctx[self.invenio_context_path]
+        self.invenio_ctx = deposition_data
 
     def map_metadata(self) -> None:
         """Map the harvested metadata onto the Invenio schema."""
 
         deposition_metadata = self._codemeta_to_invenio_deposition()
-        self.ctx.update(self.invenio_context_path["depositionMetadata"], deposition_metadata)
-
-        # Store a snapshot of the mapped data within the cache, useful for analysis, debugging, etc
-        with open(self.ctx.get_cache("deposit", self.platform_name, create=True), 'w') as invenio_json:
-            json.dump(deposition_metadata, invenio_json, indent='  ')
+        ctx = HermesContext()
+        ctx.prepare_step("deposit")
+        with ctx[self.platform_name] as deposit_ctx:
+            deposit_ctx["deposit"] = deposition_metadata
+        ctx.finalize_step("deposit")
 
     def is_initial_publication(self) -> bool:
         latest_record_id = self.invenio_ctx.get("latestRecord", {}).get("id")
@@ -426,7 +422,7 @@ def update_metadata(self) -> None:
         self.links.update(deposit["links"])
 
         _log.debug("Created new version deposit: %s", self.links["html"])
-        with open(self.ctx.get_cache('deposit', 'deposit', create=True), 'w') as deposit_file:
+        with open(self.metadata.get_cache('deposit', 'deposit', create=True), 'w') as deposit_file:
             json.dump(deposit, deposit_file, indent=4)
 
     def delete_artifacts(self) -> None:
@@ -505,7 +501,7 @@ def _codemeta_to_invenio_deposition(self) -> dict:
         differences between Invenio-based platforms.
         """
 
-        metadata = self.ctx["codemeta"]
+        metadata = self.metadata
         license = self.invenio_ctx["license"]
         communities = self.invenio_ctx["communities"]
         access_right = self.invenio_ctx["access_right"]
@@ -520,7 +516,7 @@ def _codemeta_to_invenio_deposition(self) -> dict:
                     "affiliation": author.get("affiliation", {"legalName": None}).get("legalName"),
                     # Invenio wants "family, given". author.get("name") might not have this format.
                     "name": f"{author.get('familyName')}, {author.get('givenName')}"
-                    if author.get("familyName") and author.get("givenName")
+                    if "familyName" in author and "givenName" in author
                     else author.get("name"),
                     # Invenio expects the ORCID without the URL part
                     "orcid": author.get("@id", "").replace("https://orcid.org/", "") or None,
@@ -538,7 +534,7 @@ def _codemeta_to_invenio_deposition(self) -> dict:
                     "affiliation": contributor.get("affiliation", {"legalName": None}).get("legalName"),
                     # Invenio wants "family, given". contributor.get("name") might not have this format.
                     "name": f"{contributor.get('familyName')}, {contributor.get('givenName')}"
-                    if contributor.get("familyName") and contributor.get("givenName")
+                    if "familyName" in contributor and "givenName" in contributor
                     else contributor.get("name"),
                     # Invenio expects the ORCID without the URL part
                     "orcid": contributor.get("@id", "").replace("https://orcid.org/", "") or None,
@@ -604,15 +600,15 @@ def _get_license_identifier(self) -> t.Optional[str]:
 
         If no license is configured, ``None``  will be returned.
         """
-        license_url = self.ctx["codemeta"].get("license")
+        license_url = self.metadata["license"]
         return self.resolver.resolve_license_id(license_url)
 
     def _get_community_identifiers(self):
         """Get Invenio community identifiers from config.
 
         This function gets the communities to be used for the deposition on an Invenio-based
         site from the config and checks their validity against the site's API. If one of the
-        identifiers can not be found on the site, a :class:`HermesMisconfigurationError` is
+        identifiers can not be found on the site, a :class:`MisconfigurationError` is
         raised.
         """
 

src/hermes/error.py

@@ -4,5 +4,5 @@
 
 # SPDX-FileContributor: David Pape
 
-class HermesMisconfigurationError(Exception):
+class MisconfigurationError(Exception):
     pass

src/hermes/model/api.py

@@ -1,10 +1,29 @@
+from hermes.model.context_manager import HermesContext, HermesContexError
 from hermes.model.types import ld_dict
-
 from hermes.model.types.ld_context import ALL_CONTEXTS
+from hermes.model.types.ld_dict import bundled_loader
 
 
 class SoftwareMetadata(ld_dict):
 
     def __init__(self, data: dict = None, extra_vocabs: dict[str, str] = None) -> None:
         ctx = ALL_CONTEXTS + [{**extra_vocabs}] if extra_vocabs is not None else ALL_CONTEXTS
         super().__init__([ld_dict.from_dict(data, context=ctx).data_dict if data else {}], context=ctx)
+
+    @classmethod
+    def load_from_cache(cls, ctx: HermesContext, source: str) -> "SoftwareMetadata":
+        with ctx[source] as cache:
+            try:
+                return SoftwareMetadata(cache["codemeta"])
+            except Exception:
+                pass
+            try:
+                context = cache["context"]["@context"]
+                data = SoftwareMetadata()
+                data.active_ctx = data.ld_proc.initial_ctx(context, {"documentLoader": bundled_loader})
+                data.context = context
+                for key, value in cache["expanded"][0]:
+                    data[key] = value
+                return data
+            except Exception as e:
+                raise HermesContexError("There is no (valid) data stored in the cache.") from e