You've already forked akaunting-py
127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
import requests
|
|
from requests.auth import HTTPBasicAuth
|
|
from setuptools_scm import get_version
|
|
|
|
from akauntingpy import exceptions
|
|
from akauntingpy.helpers import *
|
|
|
|
__version__ = get_version(root='..', relative_to=__file__)
|
|
|
|
|
|
class Client(object):
|
|
"""
|
|
Akaunting interface.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
url,
|
|
username,
|
|
password,
|
|
company_id,
|
|
currency_code="NZD",
|
|
currency_rate="1.0"):
|
|
"""
|
|
Create a new instance.
|
|
Args:
|
|
url (str): The URL to the Akaunting api.
|
|
username (str): The username of the Akaunting credentials.
|
|
password (str): The password of the Akaunting credentials.
|
|
"""
|
|
self.url = url
|
|
self.authentication = HTTPBasicAuth(username, password)
|
|
self.headers = {
|
|
'User-Agent': 'AkauntingPy v' + __version__,
|
|
}
|
|
self.company_id = company_id
|
|
self.default_params = {'company_id': company_id}
|
|
self.currency_code = currency_code
|
|
self.currency_rate = currency_rate
|
|
|
|
def call(self, method="get", endpoint=None,
|
|
**params) -> dict():
|
|
response = None
|
|
|
|
response = requests.request(method=method,
|
|
url=self.url + "/" + endpoint,
|
|
headers=self.headers,
|
|
auth=self.authentication,
|
|
params=MergeDict(self.default_params, params)
|
|
)
|
|
|
|
print(response)
|
|
response_ = response.json()
|
|
print(response_)
|
|
|
|
if response.status_code == 401:
|
|
raise exceptions.MissingPermission(response_['message'])
|
|
elif response.status_code == 422:
|
|
errors = []
|
|
for key in response_['errors']:
|
|
errors.append(RemoveFromString(["<strong>", "</strong>"],
|
|
response_['errors'][key][0])
|
|
+ " (" + key + ")")
|
|
raise exceptions.InvalidData(response_['message'] +
|
|
"\n" +
|
|
"\n".join(errors))
|
|
# elif response.status_code != 200 and response.status_code != 201:
|
|
# raise exceptions.Error(response_['message'])
|
|
return response_
|
|
|
|
def ping(self):
|
|
data = self.call(endpoint="ping")
|
|
return data
|
|
|
|
def get_accounts(self, **params):
|
|
data = self.call(endpoint="accounts", **params)
|
|
|
|
if params.get('search', False):
|
|
# Check if there is an account returned
|
|
if data['meta']['pagination'].get('count') == 0:
|
|
# No account found
|
|
raise exceptions.AccountNotFound("Sorry, account not found matching search parameters: %s".format(
|
|
params.get('search')
|
|
))
|
|
|
|
return data['data']
|
|
|
|
def create_transaction(self,
|
|
type='income', # Payment method type
|
|
account_id=None, # Account ID to assign
|
|
category_id=None, # Category ID to assign
|
|
contact_id=None, # Contact ID/Client to assign
|
|
description=None, # Description
|
|
paid_at=None, # Date of expense/transfer or income
|
|
reference=None, # Reference for the payment
|
|
payment_method=None, # Payment method
|
|
currency_code=None, # Currency code
|
|
currency_rate=None, # Currency rate
|
|
amount=None, # Amount received/paid
|
|
**params # Any additional parameters
|
|
):
|
|
|
|
if currency_code is None:
|
|
# Set default value from class
|
|
currency_code = self.currency_code
|
|
if currency_rate is None:
|
|
# Set default value from class
|
|
currency_rate = self.currency_rate
|
|
|
|
data = self.call(endpoint="transactions",
|
|
method="POST",
|
|
search="type:" + type,
|
|
type=type,
|
|
account_id=account_id,
|
|
category_id=category_id,
|
|
paid_at=paid_at,
|
|
contact_id=contact_id,
|
|
payment_method=payment_method,
|
|
reference=reference,
|
|
currency_code=currency_code,
|
|
currency_rate=currency_rate,
|
|
amount=amount,
|
|
description=description,
|
|
**params
|
|
)
|
|
return data
|