219 lines
7.8 KiB
Python
219 lines
7.8 KiB
Python
# -*- coding: utf-8; -*-
|
|
################################################################################
|
|
#
|
|
# Rattail -- Retail Software Framework
|
|
# Copyright © 2010-2022 Lance Edgar
|
|
#
|
|
# This file is part of Rattail.
|
|
#
|
|
# Rattail is free software: you can redistribute it and/or modify it under the
|
|
# terms of the GNU General Public License as published by the Free Software
|
|
# Foundation, either version 3 of the License, or (at your option) any later
|
|
# version.
|
|
#
|
|
# Rattail is distributed in the hope that it will be useful, but WITHOUT ANY
|
|
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
|
|
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
|
|
# details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License along with
|
|
# Rattail. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
################################################################################
|
|
"""
|
|
Harvest Web API
|
|
"""
|
|
|
|
import requests
|
|
|
|
|
|
class HarvestWebAPI(object):
|
|
"""
|
|
Generic web API object.
|
|
"""
|
|
|
|
def __init__(self, base_url=None, access_token=None, account_id=None,
|
|
user_agent=None):
|
|
self.base_url = base_url or 'https://api.harvestapp.com/v2/'
|
|
self.base_url = self.base_url.rstrip('/')
|
|
self.access_token = access_token
|
|
self.account_id = account_id
|
|
self.user_agent = user_agent
|
|
if not self.user_agent:
|
|
raise ValueError("Must provide `user_agent` when creating API instance")
|
|
|
|
def _request(self, request_method, api_method, params=None):
|
|
"""
|
|
Perform a request for the given API method, and return the response.
|
|
"""
|
|
api_method = api_method.lstrip('/')
|
|
headers = {
|
|
'Authorization': 'Bearer {}'.format(self.access_token),
|
|
'Harvest-Account-Id': self.account_id,
|
|
'User-Agent': self.user_agent,
|
|
}
|
|
if request_method == 'GET':
|
|
response = requests.get('{}/{}'.format(self.base_url, api_method),
|
|
headers=headers, params=params)
|
|
elif request_method == 'POST':
|
|
response = requests.post('{}/{}'.format(self.base_url, api_method),
|
|
headers=headers, params=params)
|
|
elif request_method == 'PATCH':
|
|
response = requests.patch('{}/{}'.format(self.base_url, api_method),
|
|
headers=headers, params=params)
|
|
else:
|
|
raise NotImplementedError("unknown request method: {}".format(
|
|
request_method))
|
|
response.raise_for_status()
|
|
return response
|
|
|
|
def get(self, api_method, params=None):
|
|
"""
|
|
Perform a GET request for the given API method, and return the response.
|
|
"""
|
|
return self._request('GET', api_method, params=params)
|
|
|
|
def post(self, api_method, params=None):
|
|
"""
|
|
Perform a POST request for the given API method, and return the response.
|
|
"""
|
|
return self._request('POST', api_method, params=params)
|
|
|
|
def patch(self, api_method, params=None):
|
|
"""
|
|
Perform a PATCH request for the given API method, and return the response.
|
|
"""
|
|
return self._request('PATCH', api_method, params=params)
|
|
|
|
def get_company(self):
|
|
"""
|
|
Retrieves the company for the currently authenticated user.
|
|
|
|
https://help.getharvest.com/api-v2/company-api/company/company/#retrieve-a-company
|
|
"""
|
|
response = self.get('/company')
|
|
return response.json()
|
|
|
|
def get_users(self, **kwargs):
|
|
"""
|
|
Retrieve all users. Any kwargs are passed on as URL query
|
|
string parameters.
|
|
|
|
https://help.getharvest.com/api-v2/users-api/users/users/#list-all-users
|
|
"""
|
|
response = self.get('/users', params=kwargs)
|
|
return response.json()
|
|
|
|
def get_clients(self, **kwargs):
|
|
"""
|
|
Retrieve all clients. Any kwargs are passed on as URL query string
|
|
parameters.
|
|
|
|
https://help.getharvest.com/api-v2/clients-api/clients/clients/#list-all-clients
|
|
"""
|
|
response = self.get('/clients', params=kwargs)
|
|
return response.json()
|
|
|
|
def get_projects(self, **kwargs):
|
|
"""
|
|
Retrieve all projects. Any kwargs are passed on as URL query string
|
|
parameters.
|
|
|
|
https://help.getharvest.com/api-v2/projects-api/projects/projects/#list-all-projects
|
|
"""
|
|
response = self.get('/projects', params=kwargs)
|
|
data = response.json()
|
|
projects = data['projects']
|
|
while data['next_page']:
|
|
|
|
kw = dict(kwargs)
|
|
kw['page'] = data['next_page']
|
|
response = self.get('/projects', params=kw)
|
|
data = response.json()
|
|
projects.extend(data['projects'])
|
|
|
|
return projects
|
|
|
|
def get_tasks(self, **kwargs):
|
|
"""
|
|
Retrieve all tasks. Any kwargs are passed on as URL query string
|
|
parameters.
|
|
|
|
https://help.getharvest.com/api-v2/tasks-api/tasks/tasks/#list-all-tasks
|
|
"""
|
|
response = self.get('/tasks', params=kwargs)
|
|
return response.json()
|
|
|
|
def get_time_entries(self, **kwargs):
|
|
"""
|
|
List all time entries. Any kwargs are passed on as URL query string
|
|
parameters.
|
|
|
|
https://help.getharvest.com/api-v2/timesheets-api/timesheets/time-entries/#list-all-time-entries
|
|
"""
|
|
response = self.get('/time_entries', params=kwargs)
|
|
data = response.json()
|
|
entries = data['time_entries']
|
|
while data['next_page']:
|
|
|
|
kw = dict(kwargs)
|
|
kw['page'] = data['next_page']
|
|
response = self.get('/time_entries', params=kw)
|
|
data = response.json()
|
|
entries.extend(data['time_entries'])
|
|
|
|
return entries
|
|
|
|
def get_time_entry(self, time_entry_id):
|
|
"""
|
|
Retrieve a time entry.
|
|
|
|
https://help.getharvest.com/api-v2/timesheets-api/timesheets/time-entries/#retrieve-a-time-entry
|
|
"""
|
|
response = self.get('/time_entries/{}'.format(time_entry_id))
|
|
return response.json()
|
|
|
|
def create_time_entry(self, **kwargs):
|
|
"""
|
|
Create a new time entry. All kwargs are passed on as POST parameters.
|
|
|
|
https://help.getharvest.com/api-v2/timesheets-api/timesheets/time-entries/#create-a-time-entry-via-duration
|
|
https://help.getharvest.com/api-v2/timesheets-api/timesheets/time-entries/#create-a-time-entry-via-start-and-end-time
|
|
"""
|
|
required = ('project_id', 'task_id', 'spent_date')
|
|
for key in required:
|
|
if key not in kwargs or not kwargs[key]:
|
|
raise ValueError("must provide all of: {}".format(', '.join(required)))
|
|
response = self.post('/time_entries', params=kwargs)
|
|
return response.json()
|
|
|
|
# TODO: deprecate / remove this
|
|
put_time_entry = create_time_entry
|
|
|
|
def stop_time_entry(self, time_entry_id):
|
|
"""
|
|
Stop a running time entry.
|
|
|
|
https://help.getharvest.com/api-v2/timesheets-api/timesheets/time-entries/#stop-a-running-time-entry
|
|
"""
|
|
response = self.patch('/time_entries/{}/stop'.format(time_entry_id))
|
|
return response.json()
|
|
|
|
def update_time_entry(self, time_entry_id, **kwargs):
|
|
"""
|
|
Update a time entry.
|
|
|
|
https://help.getharvest.com/api-v2/timesheets-api/timesheets/time-entries/#update-a-time-entry
|
|
"""
|
|
response = self.patch('/time_entries/{}'.format(time_entry_id), params=kwargs)
|
|
return response.json()
|
|
|
|
|
|
def make_harvest_webapi(config):
|
|
access_token = config.require('harvest', 'api.access_token')
|
|
account_id = config.require('harvest', 'api.account_id')
|
|
user_agent = config.require('harvest', 'api.user_agent')
|
|
return HarvestWebAPI(access_token=access_token,
|
|
account_id=account_id,
|
|
user_agent=user_agent)
|