fix: revert generic writer and define OpenAIRE awards writer logic & fix: exclude unknown fields when updating awards with subjects & 📦 release: v6.4.1

CHANGES.rst

@@ -8,6 +8,11 @@
 Changes
 =======
 
+Version v6.4.1 (released 2024-10-15)
+
+- fix: exclude unknown fields when updating awards with subjects
+- fix: revert generic writer and define OpenAIRE awards writer logic
+
 Version v6.4.0 (released 2024-10-15)
 
 - jobs: add import awards OpenAIRE; Update CORDIS

invenio_vocabularies/__init__.py

@@ -10,6 +10,6 @@
 
 from .ext import InvenioVocabularies
 
-__version__ = "6.4.0"
+__version__ = "6.4.1"
 
 __all__ = ("__version__", "InvenioVocabularies")

invenio_vocabularies/contrib/affiliations/datastreams.py

@@ -9,8 +9,11 @@
 
 """Affiliations datastreams, transformers, writers and readers."""
 
+from copy import deepcopy
+
 from flask import current_app
 
+from ...datastreams import StreamEntry
 from ...datastreams.errors import TransformerError, WriterError
 from ...datastreams.transformers import BaseTransformer
 from ...datastreams.writers import ServiceWriter
@@ -96,15 +99,24 @@ def _entry_id(self, entry):
         """Get the id from an entry."""
         return entry["id"]
 
-    def write(self, stream_entry, *args, **kwargs):
-        """Writes the input entry using a given service."""
-        entry = stream_entry.entry
-
-        return super().write(stream_entry, *args, **kwargs)
-
-    def write_many(self, stream_entries, *args, **kwargs):
-        """Writes the input entries using a given service."""
-        return super().write_many(stream_entries, *args, **kwargs)
+    def _do_update(self, entry):
+        vocab_id = self._entry_id(entry)
+        current = self._resolve(vocab_id)
+        updated = deepcopy(current.to_dict())
+
+        if "identifiers" in entry:
+            # For each new identifier
+            for new_identifier in entry["identifiers"]:
+                # Either find an existing identifier with the same scheme and update the "identifier" value
+                for existing_identifier in updated["identifiers"]:
+                    if existing_identifier["scheme"] == new_identifier["scheme"]:
+                        existing_identifier["identifier"] = new_identifier["identifier"]
+                        break
+                # Or add the new identifier to the list of identifiers
+                else:
+                    updated["identifiers"].append(new_identifier)
+
+        return StreamEntry(self._service.update(self._identity, vocab_id, updated))
 
 
 VOCABULARIES_DATASTREAM_READERS = {}

invenio_vocabularies/contrib/subjects/schema.py

@@ -13,7 +13,7 @@
 from functools import partial
 
 from invenio_i18n import get_locale
-from marshmallow import Schema, fields, pre_load
+from marshmallow import EXCLUDE, Schema, fields, pre_load
 from marshmallow_utils.fields import IdentifierSet, SanitizedUnicode
 from marshmallow_utils.schemas import IdentifierSchema
 
@@ -59,6 +59,14 @@ def add_subject_from_title(self, data, **kwargs):
 class SubjectRelationSchema(ContribVocabularyRelationSchema):
     """Schema to define an optional subject relation in another schema."""
 
+    # If re-running an OpenAIRE awards update on existing awards which already have subjects,
+    # the subject entries will contains `scheme` and `props`, which are unknown since they are `dump_only`.
+    # This makes the update exclude unknown field and go through with the update.
+    class Meta:
+        """Metadata class."""
+
+        unknown = EXCLUDE
+
     ftf_name = "subject"
     parent_field_name = "subjects"
     subject = SanitizedUnicode()

invenio_vocabularies/datastreams/writers.py

@@ -87,23 +87,8 @@ def _resolve(self, id_):
     def _do_update(self, entry):
         vocab_id = self._entry_id(entry)
         current = self._resolve(vocab_id)
-        combined_dict = current.to_dict()
-
-        # Update fields from entry
-        for key, value in entry.items():
-            if key in combined_dict:
-                if isinstance(combined_dict[key], list) and isinstance(value, list):
-                    combined_dict[key].extend(
-                        item for item in value if item not in combined_dict[key]
-                    )
-                else:
-                    combined_dict[key] = value
-            else:
-                combined_dict[key] = value
-
-        return StreamEntry(
-            self._service.update(self._identity, vocab_id, combined_dict)
-        )
+        updated = dict(current.to_dict(), **entry)
+        return StreamEntry(self._service.update(self._identity, vocab_id, updated))
 
     def write(self, stream_entry, *args, **kwargs):
         """Writes the input entry using a given service."""