quay/endpoints/api/repositorynotification.py

""" List, create and manage repository events/notifications. """

import json

from flask import request

from app import notification_queue
from endpoints.api import (RepositoryParamResource, nickname, resource, require_repo_admin,
                           log_action, validate_json_request, NotFound, request_error,
                           path_param)
from endpoints.notificationevent import NotificationEvent
from endpoints.notificationmethod import (NotificationMethod,
                                          CannotValidateNotificationMethodException)
from endpoints.notificationhelper import build_notification_data
from data import model


def notification_view(note):
  config = {}
  try:
    config = json.loads(note.config_json)
  except:
    config = {}

  return {
    'uuid': note.uuid,
    'event': note.event.name,
    'method': note.method.name,
    'config': config
  }


@resource('/v1/repository/<repopath:repository>/notification/')
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
class RepositoryNotificationList(RepositoryParamResource):
  """ Resource for dealing with listing and creating notifications on a repository. """
  schemas = {
    'NotificationCreateRequest': {
      'type': 'object',
      'description': 'Information for creating a notification on a repository',
      'required': [
        'event',
        'method',
        'config'
      ],
      'properties': {
        'event': {
          'type': 'string',
          'description': 'The event on which the notification will respond',
        },
        'method': {
          'type': 'string',
          'description': 'The method of notification (such as email or web callback)',
        },
        'config': {
          'type': 'object',
          'description': 'JSON config information for the specific method of notification'
        }
      }
    },
  }

  @require_repo_admin
  @nickname('createRepoNotification')
  @validate_json_request('NotificationCreateRequest')
  def post(self, namespace, repository):
    """ Create a new notification for the specified repository. """
    repo = model.repository.get_repository(namespace, repository)
    parsed = request.get_json()

    method_handler = NotificationMethod.get_method(parsed['method'])
    if not method_handler:
      raise request_error(message='Unknown method')

    try:
      method_handler.validate(repo, parsed['config'])
    except CannotValidateNotificationMethodException as ex:
      raise request_error(message=ex.message)

    new_notification = model.notification.create_repo_notification(repo, parsed['event'],
                                                             parsed['method'], parsed['config'])

    resp = notification_view(new_notification)
    log_action('add_repo_notification', namespace,
               {'repo': repository, 'notification_id': new_notification.uuid,
                'event': parsed['event'], 'method': parsed['method']},
               repo=repo)
    return resp, 201

  @require_repo_admin
  @nickname('listRepoNotifications')
  def get(self, namespace, repository):
    """ List the notifications for the specified repository. """
    notifications = model.notification.list_repo_notifications(namespace, repository)
    return {
      'notifications': [notification_view(n) for n in notifications]
    }


@resource('/v1/repository/<repopath:repository>/notification/<uuid>')
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
@path_param('uuid', 'The UUID of the notification')
class RepositoryNotification(RepositoryParamResource):
  """ Resource for dealing with specific notifications. """
  @require_repo_admin
  @nickname('getRepoNotification')
  def get(self, namespace, repository, uuid):
    """ Get information for the specified notification. """
    try:
      found = model.notification.get_repo_notification(uuid)
    except model.InvalidNotificationException:
      raise NotFound()

    if (found.repository.namespace_user.username != namespace or
        found.repository.name != repository):
      raise NotFound()

    return notification_view(found)

  @require_repo_admin
  @nickname('deleteRepoNotification')
  def delete(self, namespace, repository, uuid):
    """ Deletes the specified notification. """
    deleted = model.notification.delete_repo_notification(namespace, repository, uuid)
    log_action('delete_repo_notification', namespace,
               {'repo': repository, 'notification_id': uuid,
                'event': deleted.event.name, 'method': deleted.method.name},
               repo=model.repository.get_repository(namespace, repository))

    return 'No Content', 204


@resource('/v1/repository/<repopath:repository>/notification/<uuid>/test')
@path_param('repository', 'The full path of the repository. e.g. namespace/name')
@path_param('uuid', 'The UUID of the notification')
class TestRepositoryNotification(RepositoryParamResource):
  """ Resource for queuing a test of a notification. """
  @require_repo_admin
  @nickname('testRepoNotification')
  def post(self, namespace, repository, uuid):
    """ Queues a test notification for this repository. """
    try:
      test_note = model.notification.get_repo_notification(uuid)
    except model.InvalidNotificationException:
      raise NotFound()

    if (test_note.repository.namespace_user.username != namespace or
        test_note.repository.name != repository):
      raise NotFound()

    event_info = NotificationEvent.get_event(test_note.event.name)
    sample_data = event_info.get_sample_data(repository=test_note.repository)
    notification_data = build_notification_data(test_note, sample_data)
    notification_queue.put([test_note.repository.namespace_user.username, repository,
                            test_note.event.name], json.dumps(notification_data))

    return {}