views: add signposting

invenio_rdm_records/resources/config.py

@@ -52,6 +52,7 @@
     DataCite43XMLSerializer,
     DCATSerializer,
     DublinCoreXMLSerializer,
+    FAIRSignpostingProfileLvl2Serializer,
     GeoJSONSerializer,
     MARCXMLSerializer,
     SchemaorgJSONLDSerializer,
@@ -107,6 +108,7 @@ def _bibliography_headers(obj_or_list, code, many=False):
     ),
     "application/x-bibtex": ResponseHandler(BibtexSerializer(), headers=etag_headers),
     "application/dcat+xml": ResponseHandler(DCATSerializer(), headers=etag_headers),
+    "application/linkset+json": ResponseHandler(FAIRSignpostingProfileLvl2Serializer()),
 }
 
 

invenio_rdm_records/resources/resources.py

@@ -10,6 +10,8 @@
 # it under the terms of the MIT License; see LICENSE file for more details.
 
 """Bibliographic Record Resource."""
+from functools import wraps
+
 from flask import abort, current_app, g, send_file
 from flask_cors import cross_origin
 from flask_resources import (
@@ -46,6 +48,35 @@
     IIIFManifestV2JSONSerializer,
     IIIFSequenceV2JSONSerializer,
 )
+from .urls import record_url_for
+
+
+def response_header_signposting(f):
+    """Add signposting link to view's reponse headers.
+
+    :param headers: response headers
+    :type headers: dict
+    :return: updated response headers
+    :rtype: dict
+    """
+
+    @wraps(f)
+    def inner(*args, **kwargs):
+        pid_value = resource_requestctx.view_args["pid_value"]
+        signposting_link = record_url_for(_app="api", pid_value=pid_value)
+
+        response = f(*args, **kwargs)
+        if response.status_code != 200:
+            return response
+        response.headers.update(
+            {
+                "Link": f'<{signposting_link}> ; rel="linkset" ; type="application/linkset+json"',  # noqa
+            }
+        )
+
+        return response
+
+    return inner
 
 
 class RDMRecordResource(RecordResource):
@@ -89,6 +120,7 @@ def p(route):
     @request_extra_args
     @request_read_args
     @request_view_args
+    @response_header_signposting
     @response_handler()
     def read(self):
         """Read an item."""

invenio_rdm_records/resources/serializers/__init__.py

@@ -23,6 +23,7 @@
 )
 from .marcxml import MARCXMLSerializer
 from .schemaorg import SchemaorgJSONLDSerializer
+from .signposting import FAIRSignpostingProfileLvl2Serializer
 from .ui import UIJSONSerializer
 
 __all__ = (
@@ -32,6 +33,7 @@
     "DataCite43XMLSerializer",
     "DublinCoreJSONSerializer",
     "DublinCoreXMLSerializer",
+    "FAIRSignpostingProfileLvl2Serializer",
     "GeoJSONSerializer",
     "IIIFCanvasV2JSONSerializer",
     "IIIFInfoV2JSONSerializer",

invenio_rdm_records/resources/serializers/signposting/__init__.py

@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2023 Northwestern University.
+#
+# Invenio-RDM-Records is free software; you can redistribute it and/or modify
+# it under the terms of the MIT License; see LICENSE file for more details.
+
+"""Signposting serializers."""
+
+from flask_resources import BaseListSchema, MarshmallowSerializer
+from flask_resources.serializers import JSONSerializer
+
+from .schema import FAIRSignpostingProfileLvl2Schema
+
+
+class FAIRSignpostingProfileLvl2Serializer(MarshmallowSerializer):
+    """FAIR Signposting Profile level 2 serializer."""
+
+    def __init__(self):
+        """Initialise Serializer."""
+        super().__init__(
+            format_serializer_cls=JSONSerializer,
+            object_schema_cls=FAIRSignpostingProfileLvl2Schema,
+            list_schema_cls=BaseListSchema,
+        )

invenio_rdm_records/resources/serializers/signposting/schema.py

@@ -0,0 +1,228 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2023 Northwestern University.
+#
+# Invenio-RDM-Records is free software; you can redistribute it and/or modify
+# it under the terms of the MIT License; see LICENSE file for more details.
+
+"""Signposting schemas."""
+
+import idutils
+from marshmallow import Schema, fields, missing
+
+from ...urls import download_url_for
+from ..utils import get_vocabulary_props
+
+
+class LandingPageSchema(Schema):
+    """Schema for serialization of typed links pertaining to the landing page.
+
+    Serialization input is a whole record dict projection.
+    """
+
+    anchor = fields.Method(serialize="serialize_anchor")
+    author = fields.Method(serialize="serialize_author")
+    cite_as = fields.Method(data_key="cite-as", serialize="serialize_cite_as")
+    describedby = fields.Method(serialize="serialize_describedby")
+    item = fields.Method(serialize="serialize_item")
+    license = fields.Method(serialize="serialize_license")
+    type = fields.Method(serialize="serialize_type")
+
+    def serialize_anchor(self, obj, **kwargs):
+        """Seralize to landing page URL."""
+        return obj["links"]["self_html"]
+
+    def serialize_author(self, obj, **kwargs):
+        """Serialize author(s).
+
+        For now, the first linkable identifier is taken.
+        """
+
+        def pick_linkable_id(identifiers):
+            for id_dict in identifiers:
+                url = idutils.to_url(
+                    id_dict["identifier"], id_dict["scheme"], url_scheme="https"
+                )
+                if url:
+                    return url
+                else:
+                    continue
+            return None
+
+        metadata = obj["metadata"]
+        result = [
+            {"href": pick_linkable_id(c["person_or_org"].get("identifiers", []))}
+            for c in metadata.get("creators", [])
+        ]
+        result = [r for r in result if r["href"]]
+        return result or missing
+
+    def serialize_cite_as(self, obj, **kwargs):
+        """Serialize cite-as."""
+        identifier = obj.get("pids", {}).get("doi", {}).get("identifier")
+        if not identifier:
+            return missing
+
+        url = idutils.to_url(identifier, "doi", "https")
+
+        return [{"href": url}] if url else missing
+
+    def serialize_describedby(self, obj, **kwargs):
+        """Serialize describedby."""
+        # Has to be placed here to prevent circular dependency.
+        from invenio_rdm_records.resources.config import record_serializers
+
+        result = [
+            {"href": obj["links"]["self"], "type": mimetype}
+            for mimetype in sorted(record_serializers)
+        ]
+
+        return result or missing
+
+    def serialize_item(self, obj, **kwargs):
+        """Serialize item."""
+        file_entries = obj.get("files", {}).get("entries", {})
+
+        result = [
+            {
+                "href": download_url_for(pid_value=obj["id"], filename=entry["key"]),
+                "type": entry["mimetype"],
+            }
+            for entry in file_entries.values()
+        ]
+
+        return result or missing
+
+    def serialize_license(self, obj, **kwargs):
+        """Serialize license.
+
+        Note that we provide an entry for each license (rather than just 1).
+        """
+        rights = obj["metadata"].get("rights", [])
+
+        def extract_link(right):
+            """Return link associated with right.
+
+            There are 2 types of right representations:
+             - custom
+             - controlled vocabulary
+
+            If no associated link returns None.
+            """
+            # Custom
+            if right.get("link"):
+                return right["link"]
+            # Controlled vocabulary
+            elif right.get("props"):
+                return right["props"].get("url")
+
+        result = [extract_link(right) for right in rights]
+        result = [{"href": link} for link in result if link]
+        return result or missing
+
+    def serialize_type(self, obj, **kwargs):
+        """Serialize type."""
+        resource_type = obj["metadata"]["resource_type"]
+
+        props = get_vocabulary_props(
+            "resourcetypes",
+            [
+                "props.schema.org",
+            ],
+            resource_type["id"],
+        )
+        url_schema_org = props.get("schema.org")
+
+        result = []
+        if url_schema_org:
+            result.append({"href": url_schema_org})
+        # always provide About Page
+        result.append({"href": "https://schema.org/AboutPage"})
+
+        return result
+
+
+class ContentResourceSchema(Schema):
+    """Schema for serialization of typed links pertaining to the content resource.
+
+    Serialization input is a file entry dict projection.
+
+    Passing a context={"record_dict"} to the constructor is required.
+    """
+
+    anchor = fields.Method(serialize="serialize_anchor")
+    collection = fields.Method(serialize="serialize_collection")
+
+    def serialize_anchor(self, obj, **kwargs):
+        """Serialize to download url."""
+        pid_value = self.context["record_dict"]["id"]
+        return download_url_for(pid_value=pid_value, filename=obj["key"])
+
+    def serialize_collection(self, obj, **kwargs):
+        """Serialize to record landing page url."""
+        return [
+            {
+                "href": self.context["record_dict"]["links"]["self_html"],
+                "type": "text/html",
+            }
+        ]
+
+
+class MetadataResourceSchema(Schema):
+    """Schema for serialization of typed links pertaining to the metadata resource.
+
+    Serialization input is a mimetype.
+
+    Passing a context={"record_dict"} to the constructor is required.
+    """
+
+    anchor = fields.Method(serialize="serialize_anchor")
+    describes = fields.Method(serialize="serialize_describes")
+    type = fields.Method(serialize="serialize_type")
+
+    def serialize_anchor(self, obj, **kwargs):
+        """Serialize to API url."""
+        return self.context["record_dict"]["links"]["self"]
+
+    def serialize_describes(self, obj, **kwargs):
+        """Serialize to record landing page url."""
+        return [
+            {
+                "href": self.context["record_dict"]["links"]["self_html"],
+                "type": "text/html",
+            }
+        ]
+
+    def serialize_type(self, obj, **kwargs):
+        """Serialize to mimetype i.e. obj."""
+        return obj
+
+
+class FAIRSignpostingProfileLvl2Schema(Schema):
+    """FAIR Signposting Profile Lvl 2 Schema.
+
+    See https://signposting.org/FAIR/
+    """
+
+    linkset = fields.Method(serialize="serialize_linkset")
+
+    def serialize_linkset(self, obj, **kwargs):
+        """Serialize linkset."""
+        # Has to be placed here to prevent circular dependency.
+        from invenio_rdm_records.resources.config import record_serializers
+
+        result = [LandingPageSchema().dump(obj)]
+
+        content_resource_schema = ContentResourceSchema(context={"record_dict": obj})
+        result += [
+            content_resource_schema.dump(entry)
+            for entry in obj.get("files", {}).get("entries", {}).values()
+        ]
+
+        metadata_resource_schema = MetadataResourceSchema(context={"record_dict": obj})
+        result += [
+            metadata_resource_schema.dump(mimetype)
+            for mimetype in sorted(record_serializers)
+        ]
+
+        return result