You've already forked akaunting-py
117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
import requests
|
|
import json
|
|
from requests.auth import HTTPBasicAuth
|
|
from akauntingpy import exceptions
|
|
|
|
def MergeDict(dict1, dict2):
|
|
dict2.update(dict1)
|
|
return dict2
|
|
|
|
def RemoveFromString(items, string):
|
|
for item in items:
|
|
string = string.replace(item, '')
|
|
return string
|
|
class Client(object):
|
|
"""
|
|
Akaunting interface.
|
|
"""
|
|
def __init__(
|
|
self,
|
|
url,
|
|
username,
|
|
password,
|
|
company_id):
|
|
"""
|
|
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 1.0',
|
|
}
|
|
self.company_id = company_id
|
|
self.default_params = {'company_id': company_id}
|
|
|
|
def call(self, method="get", endpoint=None,
|
|
**params) -> dict():
|
|
response = None
|
|
|
|
if params is None:
|
|
response = requests.request(method=method,
|
|
url=self.url + "/" + endpoint,
|
|
headers=self.headers,
|
|
auth=self.authentication,
|
|
params=self.default_params)
|
|
else:
|
|
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_)
|
|
# try:
|
|
# result = response_['status']
|
|
# except KeyError:
|
|
# result = response_['status_code']
|
|
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=None):
|
|
data = self.call(endpoint="accounts", params=params)
|
|
|
|
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
|
|
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
|
|
):
|
|
|
|
|
|
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,
|
|
**params
|
|
)
|
|
return data |