Progress Matchbox Assignments.

progress_matchbox_assignments.py

"""
Progress Matchbox Assignments - Automatically move completed
assignments from one role to another for the same assignee and
due date.

progress (intransitive verb):
1: to move forward, proceed
2: to develop to a higher, better, or more advanced stage

author - [email protected], 4/3/2013

usage - To be run as a command line script with one argument pointing
        to a json configuration file.

        Example config:
        {
            "program": "4ea32279a54ea7c732000002",  # ID of program
            "username": "<api username>",
            "password": "<api password>",
            "last_run": "2013-04-05 12:00:00",
            "activities": [
                {
                    "from": Reader I",
                    "to": Reader II",
                    "assignee": "<user id>",
                    "date_due": "<new due date>"
                },
            ]
        }

        The above configuration will find all 'Reader I' assignments, that
        have been completed since April 5th, 2013, and create a new 'Reader II'
        for the same application, but assigned to the user listed for "assignee"
        and with the given "date_due". The "assignee" may also be `null`,
        whereupon the assignee will be the same as the one on the completed
        assignment. If "date_due" is `null`, then the date_due will similarly
        be the same as the completed assignment. The start_date will be the
        date the script is run.

        More activity pairs may be listed.

        The output of the script is a count of the number of
        assignments successfully progressed for each activity pair.
"""


import requests
from urllib import urlencode
from datetime import datetime
from dateutil import parser
import json
from sys import argv
from os import path
from shutil import copyfile

DOMAIN = 'https://app.admitpad.com'


def login(user, pswd, program):
    "Log in with given credentials, return requests session object"

    # Set up session.
    c = requests.session()
    r = c.get('%s/login/' % DOMAIN)

    # Compile login data.
    post_data = {'username': user, 'password': pswd}
    csrf = [
        cookie for cookie in c.cookies
        if cookie.name == 'csrftoken'
    ][0].value

    post_data['csrfmiddlewaretoken'] = csrf
    c.headers.update({'Referer': '%s/login/' % DOMAIN})

    # Log in.
    r = c.post('%s/login/' % DOMAIN,
               data=post_data,
               allow_redirects=True)
    assert (r.status_code == 200)

    # Choose program.
    r = c.get(
        (('%s/dashboard/choose_program?id=' + program + '&next=/') %
            DOMAIN))
    assert (r.status_code == 200)

    return c


def resource(uri, parameters={}):
    "GET a resource from the api."
    if isinstance(parameters, dict):
        parameters.update({'format': 'json'})
        params = '?' + urlencode(parameters)
    else:
        params = ''

    r = SESSION.get('%s%s%s' % (DOMAIN, uri, params))
    if (r.status_code != 200):
        # Fail if a request fails at any point.
        raise AssertionError(
            "Failed to get resource (%d): %s%s" % (
                r.status_code, uri, params))
    return r.json()


def activity_hash():
    "Name to id look-up so config activities can be stored by name."
    activity_hash = {}
    r = resource('/api/v2/activity/')
    for activity in r['objects']:
        activity_hash[activity['name']] = activity['id']
    return activity_hash


def assignment_list(id_, since, completed):
    "Work through assignment pagination."
    params = {'completed': completed, 'activity': id_}

    # Filter out assignments that we've handled
    # already in a previous run of the script.
    if since and (completed == '1'):
        params['date_completed__gt'] = since

    r = resource('/api/v2/assignment/', params)
    assigns = r['objects']

    # Paginate through.
    while r['meta']['next']:
        r = resource(r['meta']['next'], '')
        assigns += r['objects']

    return assigns


def already_assigned(assign, search_list):
    """
    Returns True if there is an assignment in search_list with
    the same application and assignee as the given assignment.
    """
    assignee = assign['assignee']
    app = assign['application']

    for upstream_assign in search_list:
        if ((upstream_assign['assignee'] == assignee) and
                (upstream_assign['application'] == app)):
            return True
    return False


def setup_activity(name, since, completed):
    "Helper function to set up activity variables."
    id_ = ACTIVITY_HASH[name]
    list_ = assignment_list(id_, since, completed)
    return id_, list_


def progress_assignments(from_name, to_name, who, when, since=None):
    from_id, from_list = setup_activity(from_name, since, '1')
    to_id, to_list = setup_activity(to_name, since, '0')

    assignee = '/api/v2/user/%s/' % who if who else None

    count = 0
    failed = []
    for assign in from_list:
        if not already_assigned(assign, to_list):
            # Create a new assignment
            base_fields = ['application', 'date_due']
            new = {}
            for field in base_fields:
                new[field] = assign[field]

            if assignee:
                new['assignee'] = assignee
            else:
                new['assignee'] = assign['assignee']

            if when:
                new['date_due'] = when

            new['activity'] = '/api/v2/activity/%s/' % to_id
            new['start_date'] = START_DATE.isoformat()

            r = SESSION.post(
                '%s/api/v2/assignment/' % DOMAIN, data=json.dumps(new),
                headers={'content-type': 'application/json',
                         'accept': 'application/json'}
            )
            if r.status_code == 201:
                count += 1
            else:
                failed.append(assign['id'])

    # Print results per activity.
    print "Progressed %d applications from '%s' to '%s'." % (
        count, from_name, to_name)
    if failed:
        print "Unable to progress the following assignments:"
        for id_ in failed:
            print id_


if __name__ == '__main__':
    # Back up and get config file.
    if len(argv) == 2 and path.exists(argv[1]):
        copyfile(argv[1], '%s.BAK' % argv[1])
        with open(argv[1]) as file_:
            SETTINGS = json.loads(file_.read())

    else:
        raise ValueError('Could not find config file.')

    START_DATE = datetime.utcnow()
    SESSION = login(
        SETTINGS['username'],
        SETTINGS['password'],
        SETTINGS['program'])

    ACTIVITY_HASH = activity_hash()

    for progression in SETTINGS['activities']:
        from_, to, who, when = [progression.get(k) for k
                                in ['from', 'to', 'assignee', 'date_due']]

        # Make sure the date is understood and formatted correctly.
        when = parser.parse(when).isoformat()

        progress_assignments(
            from_name=from_,
            to_name=to,
            who=who,
            when=when,
            since=SETTINGS.get('last_run'))

    SETTINGS['last_run'] = START_DATE.strftime('%Y-%m-%d %H:%M:%S')

    with open(argv[1], 'w') as file_:
        file_.write(json.dumps(SETTINGS,
                               sort_keys=True,
                               indent=4,
                               separators=(',', ': ')))

• Per anavedo spec. The schools in question use the same date_due for all the assignments in a round.
• I suppose it'd be more concise that way, but I prefer using a session to repeatedly sending an auth tuple in the request every time.
• We discussed this one.
• So that the script doesn't break when (if) the format changes.
• Fair point.