Extract base class from Invenio deposit plugin

src/hermes/commands/deposit/base.py

@@ -0,0 +1,56 @@
+# SPDX-FileCopyrightText: 2023 Helmholtz-Zentrum Dresden-Rossendorf (HZDR)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# SPDX-FileContributor: David Pape
+
+import click
+
+from hermes.model.context import CodeMetaContext
+
+
+# TODO: Abstract base class?
+class BaseDepositPlugin:
+    def __init__(self, click_ctx: click.Context, ctx: CodeMetaContext) -> None:
+        self.click_ctx = click_ctx
+        self.ctx = ctx
+
+    def run(self) -> None:
+        # TODO: Decide here which of initial/new/... to run?
+        steps = [
+            "prepare",
+            "map",
+            "create_initial_version",
+            "create_new_version",
+            "update_metadata",
+            "delete_artifacts",
+            "upload_artifacts",
+            "publish",
+        ]
+
+        for step in steps:
+            getattr(self, step)()
+
+    def prepare(self) -> None:
+        pass
+
+    def map(self) -> None:
+        pass
+
+    def create_initial_version(self) -> None:
+        pass
+
+    def create_new_version(self) -> None:
+        pass
+
+    def update_metadata(self) -> None:
+        pass
+
+    def delete_artifacts(self) -> None:
+        pass
+
+    def upload_artifacts(self) -> None:
+        pass
+
+    def publish(self) -> None:
+        pass

src/hermes/commands/deposit/invenio.py

@@ -17,42 +17,22 @@
 import requests
 
 from hermes import config
+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.utils import hermes_user_agent
 
 
-# TODO: Move common functionality into base class
 # TODO: Add type annotations to aid subclass implementation
-class InvenioDepositPlugin:
+class InvenioDepositPlugin(BaseDepositPlugin):
     default_licenses_api_path = "api/licenses"
     default_communities_api_path = "api/communities"
     default_depositions_api_path = "api/deposit/depositions"
 
-    def __init__(self, click_ctx: click.Context, ctx: CodeMetaContext) -> None:
-        self.click_ctx = click_ctx
-        self.ctx = ctx
 
-    def run(self):
-        # TODO: Decide here which of initial/new/... to run?
-        steps = [
-            "prepare",
-            "map",
-            "create_initial_version",
-            "create_new_version",
-            "update_metadata",
-            "delete_artifacts",
-            "upload_artifacts",
-            "publish",
-        ]
-
-        for step in steps:
-            getattr(self, step)()
-
-
-    def prepare(self):
+    def prepare(self) -> None:
         """Prepare the deposition on an Invenio-based platform.
 
         In this function we do the following:
@@ -105,7 +85,7 @@ def prepare(self):
         self.ctx.update(invenio_path["access_conditions"], access_conditions)
 
 
-    def map(self):
+    def map(self) -> None:
         """Map the harvested metadata onto the Invenio schema."""
 
         deposition_metadata = _codemeta_to_invenio_deposition(self.ctx)
@@ -118,7 +98,7 @@ def map(self):
             json.dump(deposition_metadata, invenio_json, indent='  ')
 
 
-    def create_initial_version(self):
+    def create_initial_version(self) -> None:
         """Create an initial version of a publication.
 
         If a previous publication exists, this function does nothing, leaving the work for
@@ -169,7 +149,7 @@ def create_initial_version(self):
         self.ctx.update(invenio_path["links"]["publish"], deposit["links"]["publish"])
 
 
-    def create_new_version(self):
+    def create_new_version(self) -> None:
         """Create a new version of an existing publication.
 
         If no previous publication exists, this function does nothing because
@@ -214,7 +194,7 @@ def create_new_version(self):
         self.ctx.update(invenio_path["links"]["latestDraft"], old_deposit['links']['latest_draft'])
 
 
-    def update_metadata(self):
+    def update_metadata(self) -> None:
         """Update the metadata of a draft.
 
         If no draft is found in the context, it is assumed that no metadata has to be
@@ -254,7 +234,7 @@ def update_metadata(self):
         self.ctx.update(invenio_path["links"]["publish"], deposit["links"]["publish"])
 
 
-    def delete_artifacts(self):
+    def delete_artifacts(self) -> None:
         """Delete existing file artifacts.
 
         This is done so that files which existed in an earlier publication but don't exist
@@ -265,7 +245,7 @@ def delete_artifacts(self):
         pass
 
 
-    def upload_artifacts(self):
+    def upload_artifacts(self) -> None:
         """Upload file artifacts to the deposit.
 
         We'll use the bucket API rather than the files API as it supports file sizes above
@@ -302,7 +282,7 @@ def upload_artifacts(self):
             # file_resource = response.json()
 
 
-    def publish(self):
+    def publish(self) -> None:
         """Publish the deposited record.
 
         This is done by doing a POST request to the publication URL stored in the context at

src/hermes/commands/workflow.py

@@ -16,6 +16,7 @@
 import click
 
 from hermes import config
+from hermes.commands.deposit.base import BaseDepositPlugin
 from hermes.error import MisconfigurationError
 from hermes.model.context import HermesContext, HermesHarvestContext, CodeMetaContext
 from hermes.model.errors import MergeError
@@ -210,8 +211,7 @@ def deposit(click_ctx: click.Context, initial, auth_token, file):
         click_ctx.exit(1)
 
     # TODO: Could this raise an exception?
-    # TODO: Add "BaseDepositionPlugin" as type annotation
-    deposit_plugin_class = ep.load()
+    deposit_plugin_class: BaseDepositPlugin = ep.load()
     deposit_plugin = deposit_plugin_class(click_ctx, ctx)
 
     try: