akaunting-py/akauntingpy/api.py

import requests
gimport logging
from requests.auth import HTTPBasicAuth

from akauntingpy import exceptions
from akauntingpy.helpers import *

__version__ = "1.0.5"

logger = logging.getLogger(__name__)


class Client(object):
    """
    Akaunting interface.
    """

    def __init__(
            self,
            url,
            username,
            password,
            company_id,
            ssl_verify=True,
            currency_code="NZD",
            currency_rate="1.0"):
        """
        Create a new instance.
        Args:
            url (str): The URL to the Akaunting api. ** required **
            username (str): The username of the Akaunting credentials. ** required **
            password (str): The password of the Akaunting credentials. ** required **
            company_id (str): The company ID from Akaunting
            currency_code (str): The currency code.  default is NZD
            currency_rate (str): The currency rate. default is "1.0"
        """
        self.url = url
        self.ssl_verify = ssl_verify
        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 = requests.request(method=method,
                                    url=self.url + "/" + endpoint,
                                    headers=self.headers,
                                    auth=self.authentication,
                                    params=MergeDict(self.default_params, params),
                                    verify=self.ssl_verify
                                    )

        response_ = response.json()

        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'])
        elif response.status_code == 429:
            # We hit the maximum requests
            raise exceptions.TooManyAttempts(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
            try:
                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')
                    ))
            except KeyError as e:
                # New API 3.0
                if data['meta']['total'] == 0:
                    raise exceptions.AccountNotFound("Sorry, account not found matching search parameters: %s".format(
                        params.get('search')
                    ))

        return data['data']

    def get_contact(self, **params):
        data = self.call(endpoint="contacts", **params)
        print(data)
        if params.get('search', False):
            try:
                # Check if there is an account returned
                if data['meta']['pagination'].get('count') == 0:
                    # No account found
                    raise exceptions.AccountNotFound("Sorry, contact not found matching search parameters: %s".format(
                        params.get('search')
                    ))
            except KeyError as e:
                # New API 3.0
                if data['meta']['total'] == 0:
                    raise exceptions.AccountNotFound("Sorry, contact not found matching search parameters: %s".format(
                        params.get('search')
                    ))

        return data['data']

    def create_transaction(self,
                           transaction_type='income',  # Payment method type
                           account_id=None,  # Account ID to assign
                           number="NULL",  # Transaction number
                           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:" + transaction_type,
                         number=number,
                         type=transaction_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=EnsurePositiveInteger(amount),
                         description=description,
                         **params
                         )
        return data

    def create_transfer(self,
                        from_account_id=None,  # Account ID to create transfer from
                        to_account_id=None,  # Account ID to create transfer to
                        transferred_at=None,  # Date of expense/transfer or income
                        payment_method="Bank Transfer",  # Payment method
                        amount=None,  # Amount received/paid
                        **params  # Any additional parameters
                        ):
        logger.info("Transfer called with parameters")
        logger.info("from_account_id: %s", from_account_id)
        logger.info("to_account_id: %s", to_account_id)
        logger.info("transferred_at: %s", transferred_at)
        logger.info("payment_method: %s", payment_method)
        logger.info("amount: %s", amount)

        data = self.call(endpoint="transfers",
                         method="POST",
                         from_account_id=from_account_id,
                         to_account_id=to_account_id,
                         transferred_at=transferred_at,
                         payment_method=payment_method,
                         amount=EnsurePositiveInteger(amount),
                         **params
                         )
        return data