quay/endpoints/api/webhook.py

from flask import request, url_for
from flask.ext.restful import abort

from endpoints.api import (RepositoryParamResource, nickname, resource, require_repo_admin,
                           log_action, validate_json_request)
from data import model


def webhook_view(webhook):
  return {
    'public_id': webhook.public_id,
    'parameters': json.loads(webhook.parameters),
  }


@resource('/v1/repository/<path:repository>/webhook/')
class WebhookList(RepositoryParamResource):
  """ Resource for dealing with listing and creating webhooks. """
  schemas = {
    'WebhookCreateRequest': {
      'id': 'WebhookCreateRequest',
      'type': 'object',
      'description': 'Arbitrary json.',
      'required': True,
    },
  }

  @require_repo_admin
  @nickname('createWebhook')
  @validate_json_request('WebhookCreateRequest')
  def post(self, namespace, repository):
    """ Create a new webhook for the specified repository. """
    repo = model.get_repository(namespace, repository)
    webhook = model.create_webhook(repo, request.get_json())
    resp = webhook_view(webhook)
    repo_string = '%s/%s' % (namespace, repository)
    headers = {
      'Location': url_for('api_bp.get_webhook', repository=repo_string,
                          public_id=webhook.public_id),
    }
    log_action('add_repo_webhook', namespace, 
               {'repo': repository, 'webhook_id': webhook.public_id},
               repo=repo)
    return resp, 201, headers

  @require_repo_admin
  @nickname('listWebhooks')
  def get(self, namespace, repository):
    """ List the webhooks for the specified repository. """
    webhooks = model.list_webhooks(namespace, repository)
    return {
      'webhooks': [webhook_view(webhook) for webhook in webhooks]
    }


@resource('/v1/repository/<path:repository>/webhook/<public_id>')
class Webhook(RepositoryParamResource):
  """ Resource for dealing with specific webhooks. """
  @require_repo_admin
  @nickname('getWebhook')
  def get(self, namespace, repository, public_id):
    """ Get information for the specified webhook. """
    try:
      webhook = model.get_webhook(namespace, repository, public_id)
    except model.InvalidWebhookException:
      abort(404)

    return webhook_view(webhook)

  @require_repo_admin
  @nickname('deleteWebhook')
  def delete(self, namespace, repository, public_id):
    """ Delete the specified webhook. """
    model.delete_webhook(namespace, repository, public_id)
    log_action('delete_repo_webhook', namespace, 
               {'repo': repository, 'webhook_id': public_id},
               repo=model.get_repository(namespace, repository))
    return make_response('No Content', 204)