quay/endpoints/api/repositorynotification_models_interface.py

import json
from abc import ABCMeta, abstractmethod
from collections import namedtuple

from six import add_metaclass


class RepositoryNotification(
    namedtuple(
        "RepositoryNotification",
        [
            "uuid",
            "title",
            "event_name",
            "method_name",
            "config_json",
            "event_config_json",
            "number_of_failures",
        ],
    )
):
    """
    RepositoryNotification represents a notification for a repository.

    :type uuid: string
    :type event: string
    :type method: string
    :type config: string
    :type title: string
    :type event_config: string
    :type number_of_failures: int
    """

    def to_dict(self):
        try:
            config = json.loads(self.config_json)
        except ValueError:
            config = {}

        try:
            event_config = json.loads(self.event_config_json)
        except ValueError:
            event_config = {}

        return {
            "uuid": self.uuid,
            "title": self.title,
            "event": self.event_name,
            "method": self.method_name,
            "config": config,
            "event_config": event_config,
            "number_of_failures": self.number_of_failures,
        }


@add_metaclass(ABCMeta)
class RepoNotificationInterface(object):
    """
    Interface that represents all data store interactions required by the RepositoryNotification
    API.
    """

    @abstractmethod
    def create_repo_notification(
        self,
        namespace_name,
        repository_name,
        event_name,
        method_name,
        method_config,
        event_config,
        title=None,
    ):
        """

        Args:
          namespace_name: namespace of repository
          repository_name: name of repository
          event_name: name of event
          method_name: name of method
          method_config: method config, json string
          event_config: event config, json string
          title: title of the notification

        Returns:
          RepositoryNotification object

        """
        pass

    @abstractmethod
    def list_repo_notifications(self, namespace_name, repository_name, event_name=None):
        """

        Args:
          namespace_name: namespace of repository
          repository_name: name of repository
          event_name: name of event

        Returns:
          list(RepositoryNotification)
        """
        pass

    @abstractmethod
    def get_repo_notification(self, uuid):
        """

        Args:
          uuid: uuid of notification

        Returns:
          RepositoryNotification or None

        """
        pass

    @abstractmethod
    def delete_repo_notification(self, namespace_name, repository_name, uuid):
        """

        Args:
          namespace_name: namespace of repository
          repository_name: name of repository
          uuid: uuid of notification

        Returns:
          RepositoryNotification or None

        """
        pass

    @abstractmethod
    def reset_notification_number_of_failures(self, namespace_name, repository_name, uuid):
        """

        Args:
          namespace_name: namespace of repository
          repository_name: name of repository
          uuid: uuid of notification

        Returns:
          RepositoryNotification

        """
        pass

    @abstractmethod
    def queue_test_notification(self, uuid):
        """

        Args:
          uuid: uuid of notification

        Returns:
          RepositoryNotification or None

        """
        pass