/HarvestTaskSwitch

Simple Python script using Requests to update Harvest task associations.

MIT LicenseMIT

HarvestTaskSwitch

Simple Python3 script using Requests to update Harvest task associations.

Requirements
Requests pip3 install requests

Code:

from base64 import b64encode
import requests

domain = 'YOURACCOUNT' # https://YOURACCOUNT.harvestapp.com/
username = '' # This uses Basic Authentication
password = '' #
old_tasks = [7206757, 7206766]
new_task = 6868062


class harvestAuth(requests.auth.AuthBase):
    """This class sets up the authorization and JSON request headers."""
    def __init__(self, username, password):
        self.username = username
        self.password = password

    def __eq__(self, other):
        return all([
            self.username == getattr(other, 'username', None),
            self.password == getattr(other, 'password', None)
        ])

    def __ne__(self, other):
        return not self == other

    def __call__(self, r):
        r.headers['Content-type'] = 'application/json'
        r.headers['Accept'] = 'application/json'
        encode = ':'.join((self.username, self.password))
        r.headers['Authorization'] = 'Basic ' + (b64encode(encode.encode('utf8'))).decode('ascii').strip()
        return r


def caller(page, action='get', payload=None, params=None, domain=domain, username=username, password=password):
    """This function sets up an easy way to GET or POST Harvest API"""
    if (action == 'get' and payload is None):
        r = requests.get('https://' + domain + '.harvestapp.com/' + page, auth=harvestAuth(username=username, password=password))
        if (r.status_code == requests.codes.ok):
            return {'json': r.json(), 'text': r.text, 'status': r.status_code}
        else:
            return {'status': r.status_code, 'text': r.text}
    elif (action == 'post' and payload is not None and params is not None):
        r = requests.post('https://' + domain + '.harvestapp.com/' + page, auth=harvestAuth(username=username, password=password), json=payload, params=params)
        return {'status': r.status_code, 'text': r.text}
    else:
        return False


resp = caller('people')

count = 0

for person in resp['json']:
    mC = 0
    person = person['user']
    tasks = caller('people/'+str(person['id'])+'/entries?from=20000101&to=22221230')
    for task in tasks['json']:
        task = task['day_entry']
        if (task['task_id'] in old_tasks):
            resp = caller(page='daily/update/'+str(task['id']),
                          action='post',
                          payload={"notes": task['notes'], "hours": task['hours'], "project_id": task['project_id'], "task_id": str(new_task), "spent_at": task['spent_at']},
                          params={'of_user': str(person['id'])})
            if (resp['status'] == requests.codes.ok):
                mC = mC + 1
            else:
                print('Error: Could not update User: '+str(person['id'])+' Task: ' + str(task['id']))
    count = count + mC
    print('Finished ' + person['first_name'] + ' with ' + str(mC) + ' updates.')

print('Total updates: ' + str(count))