Flesh out the webworkers a bit.

data/database.py

@@ -104,11 +104,6 @@ class Repository(BaseModel):
     )
 
 
-class Webhook(BaseModel):
-  repository = ForeignKeyField(Repository)
-  parameters = TextField()
-
-
 class Role(BaseModel):
   name = CharField(index=True)
 
@@ -135,6 +130,13 @@ def random_string_generator(length=16):
   return random_string
 
 
+class Webhook(BaseModel):
+  public_id = CharField(default=random_string_generator(length=64),
+                        unique=True, index=True)
+  repository = ForeignKeyField(Repository)
+  parameters = TextField()
+
+
 class AccessToken(BaseModel):
   friendly_name = CharField(null=True)
   code = CharField(default=random_string_generator(length=64), unique=True,

data/model.py

@@ -2,6 +2,7 @@ import bcrypt
 import logging
 import dateutil.parser
 import operator
+import json
 
 from database import *
 from util.validation import *
@@ -42,6 +43,10 @@ class InvalidRepositoryBuildException(DataModelException):
   pass
 
 
+class InvalidWebhookException(DataModelException):
+  pass
+
+
 def create_user(username, password, email):
   if not validate_email(email):
     raise InvalidEmailAddressException('Invalid email address: %s' % email)
@@ -935,3 +940,25 @@ def list_repository_builds(namespace_name, repository_name,
 def create_repository_build(repo, access_token, resource_key, tag):
   return RepositoryBuild.create(repository=repo, access_token=access_token,
                                 resource_key=resource_key, tag=tag)
+
+
+def create_webhook(repo, params_obj):
+  return Webhook.create(repository=repo, parameters=json.dumps(params_obj))
+
+
+def get_webhook(namespace_name, repository_name, public_id):
+  joined = Webhook.select().join(Repository)
+  found = list(joined.where(Repository.namespace == namespace_name,
+                            Repository.name == repository_name,
+                            Webhook.public_id == public_id))
+
+  if not found:
+    raise InvalidWebhookException('No webhook found with id: %s' % public_id)
+
+  return found[0]
+
+
+def list_webhooks(namespace_name, repository_name):
+  joined = Webhook.select().join(Repository)
+  return joined.where(Repository.namespace == namespace_name,
+                      Repository.name == repository_name)

endpoints/api.py

@@ -4,7 +4,7 @@ import requests
 import urlparse
 import json
 
-from flask import request, make_response, jsonify, abort
+from flask import request, make_response, jsonify, abort, url_for
 from flask.ext.login import current_user, logout_user
 from flask.ext.principal import identity_changed, AnonymousIdentity
 from functools import wraps
@@ -799,6 +799,56 @@ def request_repo_build(namespace, repository):
   abort(403)  # Permissions denied
 
 
+def webhook_view(webhook):
+  return {
+    'public_id': webhook.public_id,
+    'parameters': json.loads(webhook.parameters),
+  }
+
+
+@app.route('/api/repository/<path:repository>/webhook/', methods=['POST'])
+@api_login_required
+@parse_repository_name
+def create_webhook(namespace, repository):
+  permission = ModifyRepositoryPermission(namespace, repository)
+  if permission.can():
+    repo = model.get_repository(namespace, repository)
+    webhook = model.create_webhook(repo, request.get_json())
+    resp = jsonify(webhook_view(webhook))
+    repo_string = '%s/%s' % (namespace, repository)
+    resp.headers['Location'] = url_for('get_webhook', repository=repo_string,
+                                       public_id=webhook.public_id)
+
+  abort(403)  # Permissions denied
+
+
+@app.route('/api/repository/<path:repository>/webhook/<public_id>',
+           methods=['GET'])
+@api_login_required
+@parse_repository_name
+def get_webhook(namespace, repository, public_id):
+  permission = ModifyRepositoryPermission(namespace, repository)
+  if permission.can():
+    webhook = model.get_webhook(namespace, repository, public_id)
+    return jsonify(webhook_view(webhook))
+
+  abort(403)  # Permission denied
+
+
+@app.route('/api/repository/<path:repository>/webhook/', methods=['GET'])
+@api_login_required
+@parse_repository_name
+def list_webhooks(namespace, repository):
+  permission = ModifyRepositoryPermission(namespace, repository)
+  if permission.can():
+    webhooks = model.list_webhooks(namespace, repository)
+    return jsonify({
+      'webhooks': [webhook_view(webhook) for webhook in webhooks]
+    })
+
+  abort(403)  # Permission denied
+
+
 @app.route('/api/filedrop/', methods=['POST'])
 @api_login_required
 def get_filedrop_url():

workers/webhookworker.py

@@ -0,0 +1,59 @@
+import logging
+import daemon
+import argparse
+import requests
+
+from data.queue import webhook_queue
+from workers.worker import Worker
+
+
+root_logger = logging.getLogger('')
+root_logger.setLevel(logging.DEBUG)
+
+FORMAT = '%(asctime)-15s - %(levelname)s - %(pathname)s - %(funcName)s - %(message)s'
+formatter = logging.Formatter(FORMAT)
+
+logger = logging.getLogger(__name__)
+
+
+class WebhookWorker(Worker):
+  def process_queue_item(self, job_details):
+    url = job_details['url']
+    payload = job_details['payload']
+
+    try:
+      resp = requests.post(url, data=payload)
+      if resp.status_code/100 != 2:
+        logger.error('%s response for webhook to url: %s' % (resp.status_code,
+                                                             url))
+        return False
+    except requests.exceptions.RequestException as ex:
+      logger.exception('Webhook was unable to be sent: %s' % ex.message)
+      return False
+
+    return True
+
+
+parser = argparse.ArgumentParser(description='Worker daemon to send webhooks')
+parser.add_argument('-D', action='store_true', default=False,
+                    help='Run the worker in daemon mode.')
+parser.add_argument('--log', default='webhooks.log',
+                    help='Specify the log file for the worker as a daemon.')
+args = parser.parse_args()
+
+
+worker = WebhookWorker(webhook_queue, poll_period_seconds=15,
+                       reservation_seconds=3600)
+
+if args.D:
+  handler = logging.FileHandler(args.log)
+  handler.setFormatter(formatter)
+  root_logger.addHandler(handler)
+  with daemon.DaemonContext(files_preserve=[handler.stream]):
+    worker.start()
+
+else:
+  handler = logging.StreamHandler()
+  handler.setFormatter(formatter)
+  root_logger.addHandler(handler)
+  worker.start()