rattail-harvest/rattail_harvest/harvest/webapi.py

170 lines
6 KiB
Python
Raw Normal View History

# -*- 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)
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 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)
return response.json()
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 put_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()