Source code for swh.coarnotify.server.handlers
# Copyright (C) 2025 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
"""Module dedicated to the handlers of COAR Notifications."""
from importlib.metadata import version
import json
from typing import Callable
from django.conf import settings
from swh.model.model import (
MetadataAuthority,
MetadataAuthorityType,
MetadataFetcher,
Origin,
RawExtrinsicMetadata,
)
from swh.storage import get_storage
from .models import InboundNotification, Statuses
from .utils import create_accept_cn, reject, send_cn, to_sorted_tuple, unprocessable
CNHandler = Callable[[InboundNotification], None]
[docs]
def get_handler(notification: InboundNotification) -> CNHandler | None:
"""Get a CN handler from its type.
The list of handlers by type is defined in the ``handlers`` dict.
Args:
notification: an inbound CN
Raises:
UnprocessableException: no handler available for cn
Returns:
A COAR Notification handler if one matches
"""
type_ = to_sorted_tuple(notification.payload["type"])
try:
return handlers[type_]
except KeyError:
error_message = f"Unable to process {', '.join(type_)} COAR Notifications"
unprocessable(notification, error_message)
return None
[docs]
def mention(notification: InboundNotification) -> None:
"""Handle a mention COAR Notification.
Validates the payload, sends the CN to the Raw Extrinsic Metadata storage and
send an Accept CN.
Args:
cn: an inbound CN
"""
# validate the payload and extract data or reject
context_data = notification.payload["context"] # describe the paper
object_data = notification.payload["object"] # describe the relationship
paper_url = context_data["id"]
if paper_url != object_data["as:subject"]:
error_message = (
f"Context id {paper_url} does not match "
f"object as:subject {object_data['as:subject']}"
)
reject(notification, error_message)
return
context_type = to_sorted_tuple(context_data["type"])
if "sorg:AboutPage" not in context_type:
error_message = "Context type does not contain sorg:AboutPage"
reject(notification, error_message)
return
origin_url = object_data["as:object"]
# Store the CN in the Raw Extrinsic Metadata storage
storage = get_storage(**settings.SWH_CONF["storage"])
metadata_fetcher = MetadataFetcher(
name="swh-coarnotify", version=version("swh-coarnotify")
)
origin = Origin(origin_url)
swhid = origin.swhid()
metadata_authority = MetadataAuthority(
type=MetadataAuthorityType.REGISTRY,
url=notification.payload["origin"]["id"],
)
metadata_object = RawExtrinsicMetadata(
target=swhid,
discovery_date=notification.created_at,
authority=metadata_authority,
fetcher=metadata_fetcher,
format="coarnotify-mention-v1",
metadata=json.dumps(notification.payload).encode(),
)
storage.metadata_authority_add([metadata_authority])
storage.metadata_fetcher_add([metadata_fetcher])
storage.raw_extrinsic_metadata_add([metadata_object])
# update cn and send a reply
notification.status = Statuses.ACCEPTED
notification.save()
accepted_cn = create_accept_cn(notification, summary=f"Stored mention for {swhid}")
send_cn(accepted_cn)
handlers = {
("Announce", "RelationshipAction"): mention,
}