subjects: mesh: add mesh subject datastreams

invenio_vocabularies/contrib/subjects/datastreams.py

@@ -12,6 +12,9 @@
 from invenio_i18n import lazy_gettext as _
 
 from ...datastreams.writers import ServiceWriter
+from .mesh.datastreams import VOCABULARIES_DATASTREAM_READERS as mesh_readers
+from .mesh.datastreams import VOCABULARIES_DATASTREAM_TRANSFORMERS as mesh_transformers
+from .mesh.datastreams import VOCABULARIES_DATASTREAM_WRITERS as mesh_writers
 
 
 class SubjectsServiceWriter(ServiceWriter):
@@ -27,14 +30,15 @@ def _entry_id(self, entry):
         return entry["id"]
 
 
-VOCABULARIES_DATASTREAM_READERS = {}
+VOCABULARIES_DATASTREAM_READERS = {**mesh_readers}
 """Subjects Data Streams readers."""
 
-VOCABULARIES_DATASTREAM_TRANSFORMERS = {}
+VOCABULARIES_DATASTREAM_TRANSFORMERS = {**mesh_transformers}
 """Subjects Data Streams transformers."""
 
 VOCABULARIES_DATASTREAM_WRITERS = {
     "subjects-service": SubjectsServiceWriter,
+    **mesh_writers,
 }
 """Subjects Data Streams writers."""
 

invenio_vocabularies/contrib/subjects/mesh/datastreams.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2022-2024 CERN.
+# Copyright (C) 2024 California Institute of Technology.
+#
+# Invenio-Vocabularies is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""MeSH subjects datastreams, transformers, writers and readers."""
+
+from invenio_vocabularies.datastreams.transformers import (
+    BaseTransformer,
+    TransformerError,
+)
+
+
+class MeshSubjectsTransformer(BaseTransformer):
+    """MeSH subjects Transformer."""
+
+    def apply(self, stream_entry, *args, **kwargs):
+        """Apply transformation on steam entry."""
+        entry_data = stream_entry.entry
+
+        # ID in MeSH data is the URL, ex. https://id.nlm.nih.gov/mesh/D000001
+        # We just want to use the ID prefixed by "mesh:""
+        try:
+            mesh_id = entry_data["id"].split("/")[-1]
+        except Exception:
+            raise TransformerError("Not a valid MeSH ID.")
+
+        entry_data["id"] = "mesh:" + mesh_id
+        return stream_entry
+
+
+VOCABULARIES_DATASTREAM_READERS = {}
+"""MeSH datastream readers."""
+
+VOCABULARIES_DATASTREAM_WRITERS = {}
+"""MeSH subject datastream writers."""
+
+VOCABULARIES_DATASTREAM_TRANSFORMERS = {"mesh-subjects": MeshSubjectsTransformer}
+"""MeSH subjects datastream transformers."""