Source code for swh.coarnotify.client

# Copyright (C) 2025 - 2026  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
"""COAR Notify clients."""

import pprint

from django.conf import settings
import requests

from swh.coarnotify.server.models import OutboundNotification




class COARNotifyClient:
    """A basic CN client."""

    def _get_inbox_url(self, notification: OutboundNotification) -> str:
        """Get target's inbox URL from the notification.

        Args:
            notification: an outbound notification

        Returns:
            an inbox URL
        """
        return notification.inbox



    def send(self, notification: OutboundNotification) -> bool:
        """Send the notification using requests.

        Args:
            notification: an outbound notification

        Raises:
            request.HTTPError: the inbox rejected our notification

        Returns:
            True if the inbox accepted the notification
        """
        inbox_url = self._get_inbox_url(notification)
        headers = {"Content-type": "application/ld+json"}
        r = requests.post(
            inbox_url,
            json=notification.payload,
            headers=headers,
            timeout=settings.CN_SEND_TIMEOUT,
        )
        r.raise_for_status()
        return True






class DevCOARNotifyClient(COARNotifyClient):
    """CN client that sends notification to a single inbox URL."""

    def _get_inbox_url(self, notification: OutboundNotification) -> str:
        """Get the dev inbox URL from the settings.

        Args:
            notification: an outbound notification

        Returns:
            the dev inbox URL
        """
        return settings.CN_INBOX_URL_OVERRIDE





class DummyCOARNotifyClient(COARNotifyClient):
    """Dummy CN client that does nothing."""



    def send(self, notification: OutboundNotification) -> bool:
        """Pretends to send the notification.

        Args:
            notification: an outbound notification

        Returns:
            Always returns True
        """
        return True






class ConsoleCOARNotifyClient(COARNotifyClient):
    """CN client that sends notification to the console."""



    def send(self, notification: OutboundNotification) -> bool:
        """Send the notification using requests.

        Args:
            notification: an outbound notification

        Raises:
            request.HTTPError: the inbox rejected our notification

        Returns:
            True if the inbox accepted the notification
        """
        pprint.pprint(notification.payload)
        print("-" * 79, "\n")
        return True