Source code for gratipay.project_review_process

# -*- coding: utf-8 -*-
from __future__ import absolute_import, division, print_function, unicode_literals

import json
import pprint
import requests
import sys

from aspen import log

from gratipay.exceptions import NoTeams
from gratipay.models.participant import Participant

class ProjectReviewProcess(object):

    def __init__(self, env, db, email_queue):
        repo = env.project_review_repo
        auth = (env.project_review_username, env.project_review_token)
        self.db = db
        self.email_queue = email_queue
        self._poster = GitHubPoster(repo, auth) if repo else ConsolePoster()

    def start(self, *teams):
        """Given team objects, kick off a review process by:

        1. creating an issue in our project review repo on GitHub, and
        2. sending an email notification to the owner of the team(s).

        It's a bug to pass in teams that don't all have the same owner.

        :return: the URL of the new review issue

        if not teams:
            raise NoTeams()
        nteams = len(teams)
        if nteams == 1:
            title = teams[0].name
        elif nteams == 2:
            title = "{} and {}".format(*[ for t in teams])
            title = "{} and {} other projects".format(teams[0].name, nteams-1)

        body = []
        team_ids = []
        owner_usernames = set()
        for team in teams:
        assert len(owner_usernames) == 1, owner_usernames
        body.extend(['', '(This application will remain open for at least a week.)'])
        data = json.dumps({'title': title, 'body': '\n'.join(body)})
        review_url ="UPDATE teams SET review_url=%s WHERE id = ANY(%s)", (review_url, team_ids))
        [team.set_attributes(review_url=review_url) for team in teams]

        owner = Participant.from_username(owner_usernames.pop())
        self.email_queue.put( owner
                            , 'project-review'
                            , review_url=team.review_url
                            , include_unsubscribe=False
                            , _user_initiated=False

        return review_url

[docs]class GitHubPoster(object): """Sends data to GitHub. """ def __init__(self, repo, auth): self.repo = repo self.api_url = "{}/issues".format(repo) self.auth = auth
[docs] def post(self, data): """POST data to GitHub and return the issue URL. """ out = '' try: r =, auth=self.auth, data=data) if r.status_code == 201: out = r.json()['html_url'] else: log(r.status_code) log(r.text) err = str(r.status_code) except: err = "eep" if not out: out = "{}/issues#error-{}".format(self.repo, err) return out
[docs]class ConsolePoster(object): """Dumps data to stdout. """ def __init__(self, fp=sys.stdout): self.fp = fp
[docs] def post(self, data): """POST data to nowhere and return a URL of lies. """ p = lambda *a, **kw: print(*a, file=self.fp) p('-'*78,) p(pprint.pformat(json.loads(data))) p('-'*78) return 'some-github-issue'