78 lines
2.6 KiB
Python
78 lines
2.6 KiB
Python
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)
|