swh.coarnotify.server.utils module#
Utils.
- swh.coarnotify.server.utils.send_cn(notification: OutboundNotification) None[source]#
Schedule the sending of a COAR Notification.
- Parameters:
notification – an OutboundNotification
- swh.coarnotify.server.utils.create_reject_cn(cn: InboundNotification) OutboundNotification[source]#
Create an Outbound CN to reply to a rejected CN.
https://coar-notify.net/specification/1.0.1/reject/
- Parameters:
cn – an inbound CN
- Returns:
an outbound CN
- swh.coarnotify.server.utils.create_accept_cn(cn: InboundNotification, summary: str | None = None) OutboundNotification[source]#
Create an Outbound CN to reply to an acceptable CN.
https://coar-notify.net/specification/1.0.1/accept/
- Parameters:
cn – an inbound notification
summary – an optional summary to include in the CN
- Returns:
an outbound CN
- swh.coarnotify.server.utils.create_unprocessable_cn(cn: InboundNotification) OutboundNotification[source]#
Create an Outbound CN to reply to an unprocessable CN.
https://coar-notify.net/specification/1.0.1/unprocessable/
- Parameters:
notification – an inbound CN
- Returns:
an outbound CN
- swh.coarnotify.server.utils.uuid_from_urn(urn: str) UUID[source]#
Extract a UUID from a URN.
- Parameters:
urn – a UUID URN (urn:uuid:xxx)
- Raises:
ValueError – URN is not a valid UUID URN
- Returns:
a uuid
- swh.coarnotify.server.utils.get_in_reply_to(graph: Graph, root_id: str) OutboundNotification | None[source]#
Get the
inReplyTonotification (if it exists).- Parameters:
graph – RDF graph
root_id – the document ID
- Returns:
The matching OutboundNotification if it exists
- swh.coarnotify.server.utils.get_notification_urn(document: list[dict]) str[source]#
Get the notification ID from an expanded JSON-LD.
- Parameters:
document – an expanded JSON-LD
- Raises:
serializers.ValidationError – invalid ID or duplicate
- Returns:
A UUID URN
- swh.coarnotify.server.utils.get_handler_from_types(graph: Graph, root_id: str) Handlers | None[source]#
Get the notification handler from its types.
For now only mentions (Announce + RelationshipAction) are handled.
- Parameters:
graph – RDF graph
root_id – the document ID
- Returns:
An handler name
- swh.coarnotify.server.utils.reject(notification: InboundNotification, error_message: str) None[source]#
Mark notification as rejected and send a Reject CN.
- Parameters:
notification – an InboundNotification
error_message – a reason for the rejection
- swh.coarnotify.server.utils.unprocessable(notification: InboundNotification, error_message: str) None[source]#
Mark notification as unprocessable and send an Unprocessable CN.
- Parameters:
notification – an InboundNotification
error_message – a reason for the rejection