""" 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//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//notification/') @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//notification//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 {}