Add celery worker and consensus tasks

Author: tedivm

bin/worker

@@ -0,0 +1,13 @@
+#!/usr/bin/env bash
+
+# Get real directory in case of symlink
+if [[ -L "${BASH_SOURCE[0]}" ]]
+then
+  DIR="$( cd "$( dirname $( readlink "${BASH_SOURCE[0]}" ) )" && pwd )"
+else
+  DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+fi
+
+source "$DIR/envvar"
+
+celery -A gitconsensusservice.celery worker --loglevel=info

gitconsensusservice/__init__.py

@@ -1,3 +1,4 @@
+from celery import Celery
 from flask import Flask
 import os
 import yaml
@@ -9,3 +10,9 @@
     if os.path.isfile(os.environ['SETTINGS']):
         with open(os.environ['SETTINGS'], 'r') as infile:
             app.config.update(yaml.load(infile.read()))
+
+if 'CELERY_BROKER' in app.config:
+    print(app.config['CELERY_BROKER'])
+    celery = Celery('gitconsensus', broker=app.config['CELERY_BROKER'])
+else:
+    celery = Celery('gitconsensus')

gitconsensusservice/jobs/consensus.py

@@ -0,0 +1,58 @@
+from gitconsensusservice import app, celery
+from services.githubapp import gh
+
+
+@celery.task
+def process_installs(synchronous=False):
+    print('Processing all installations')
+    installs = gh.get_installations()
+    for install in installs:
+        if synchronous:
+            process_installation(install, True)
+        else:
+            process_installation.delay(install)
+
+
+@celery.task
+def process_installation(installation_id, synchronous=False):
+    print('Processing installation %s' % (installation_id))
+    installation = gh.get_installation(installation_id)
+    repositories = installation.get_repositories()
+    for repository in repositories:
+        user, repo = repository.split('/')
+        process_repository(installation_id, user, repo, True)
+
+
+@celery.task
+def process_repository(installation_id, user, repo, synchronous=False):
+    print('Processing %s/%s as installation %s' % (user, repo, installation_id))
+    installation = gh.get_installation(installation_id)
+    repository = installation.get_repository(user, repo)
+
+    # has consensus rules?
+    if not repository.rules:
+        print('%s/%s does not have any consensus rules.' % (user, repo))
+        return
+
+    prs = installation.get_pr_numbers(user, repo)
+    for pr in prs:
+        if synchronous:
+            process_pull_request(installation_id, user, repo, pr)
+        else:
+            process_pull_request.delay(installation_id, user, repo, pr)
+
+
+@celery.task
+def process_pull_request(installation_id, user, repo, pull_request):
+    print('Processing %s/%s #%s as installation %s' % (user, repo, pull_request, installation_id))
+    installation = gh.get_installation(installation_id)
+    repository = installation.get_repository(user, repo)
+    request = repository.getPullRequest(pull_request)
+    if request.validate():
+        print("Merging PR#%s" % (request.number,))
+        request.vote_merge()
+    elif request.shouldClose():
+        print("Closing PR#%s" % (request.number,))
+        request.close()
+    else:
+        request.addInfoLabels()

settings.dist.yaml

@@ -2,3 +2,4 @@ DEBUG: false
 GITHUB_PRIVATE_KEY: '/absolute/path/to/private-key.pem'
 GITHUB_APP_ID: 0
 GITHUB_WEBHOOK_SECRET:
+CELERY_BROKER: 'pyamqp://guest@localhost//'