Api base

hypha.apply.projects.services.sageintacct.wrapper.api_base

ApiBase

ApiBase(dimension=None, pagesize=2000, post_legacy_method=None)

The base class for all API classes.

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def __init__(
    self,
    dimension: str = None,
    pagesize: int = 2000,
    post_legacy_method: str = None,
):
    self.__sender_id = None
    self.__sender_password = None
    self.__session_id = None
    self.__api_url = "https://api.intacct.com/ia/xml/xmlgw.phtml"
    self.__dimension = dimension
    self.__pagesize = pagesize
    self.__post_legacy_method = post_legacy_method

set_sender_id

set_sender_id(sender_id)

Set the sender id for APIs :param sender_id: sender id :return: None

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def set_sender_id(self, sender_id: str):
    """
    Set the sender id for APIs
    :param sender_id: sender id
    :return: None
    """
    self.__sender_id = sender_id

set_sender_password

set_sender_password(sender_password)

Set the sender password for APIs :param sender_password: sender id :return: None

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def set_sender_password(self, sender_password: str):
    """
    Set the sender password for APIs
    :param sender_password: sender id
    :return: None
    """
    self.__sender_password = sender_password

get_session_id

get_session_id(user_id, company_id, user_password, entity_id=None)

Sets the session id for APIs :param access_token: acceess token (JWT) :return: session id

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def get_session_id(
    self, user_id: str, company_id: str, user_password: str, entity_id: str = None
):
    """
    Sets the session id for APIs
    :param access_token: acceess token (JWT)
    :return: session id
    """

    timestamp = datetime.datetime.now()
    dict_body = {
        "request": {
            "control": {
                "senderid": self.__sender_id,
                "password": self.__sender_password,
                "controlid": timestamp,
                "uniqueid": False,
                "dtdversion": 3.0,
                "includewhitespace": False,
            },
            "operation": {
                "authentication": {
                    "login": {
                        "userid": user_id,
                        "companyid": company_id,
                        "password": user_password,
                        "locationid": entity_id,
                    }
                },
                "content": {
                    "function": {
                        "@controlid": str(uuid.uuid4()),
                        "getAPISession": None,
                    }
                },
            },
        }
    }

    response = self.__post_request(dict_body, self.__api_url)

    if response["authentication"]["status"] == "success":
        session_details = response["result"]["data"]["api"]
        self.__api_url = session_details["endpoint"]
        self.__session_id = session_details["sessionid"]

        return self.__session_id

    else:
        raise SageIntacctSDKError("Error: {0}".format(response["errormessage"]))

set_session_id

set_session_id(session_id)

Set the session id for APIs :param session_id: session id :return: None

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def set_session_id(self, session_id: str):
    """
    Set the session id for APIs
    :param session_id: session id
    :return: None
    """
    self.__session_id = session_id

format_and_send_request

format_and_send_request(data)

Format data accordingly to convert them to xml.

Parameters:

• data (dict) –

  HTTP POST body data for the wanted API.

Returns:

• –

  A response from the __post_request (dict).

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def format_and_send_request(self, data: Dict):
    """Format data accordingly to convert them to xml.

    Parameters:
        data (dict): HTTP POST body data for the wanted API.

    Returns:
        A response from the __post_request (dict).
    """

    key = next(iter(data))
    timestamp = datetime.datetime.now()

    dict_body = {
        "request": {
            "control": {
                "senderid": self.__sender_id,
                "password": self.__sender_password,
                "controlid": timestamp,
                "uniqueid": False,
                "dtdversion": 3.0,
                "includewhitespace": False,
            },
            "operation": {
                "authentication": {"sessionid": self.__session_id},
                "content": {
                    "function": {"@controlid": str(uuid.uuid4()), key: data[key]}
                },
            },
        }
    }

    response = self.__post_request(dict_body, self.__api_url)
    return response["result"]

post

post(data)

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def post(self, data: Dict):
    if self.__dimension in ("CCTRANSACTION", "EPPAYMENT"):
        return self.__construct_post_legacy_payload(data)

    return self.__construct_post_payload(data)

count

count()

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def count(self):
    get_count = {
        "query": {
            "object": self.__dimension,
            "select": {"field": "RECORDNO"},
            "pagesize": "1",
        }
    }

    response = self.format_and_send_request(get_count)
    return int(response["data"]["@totalcount"])

read_by_query

read_by_query(fields=None)

Read by Query from Sage Intacct

Parameters:

• fields (list, default: None ) –

  Get selective fields to be returned. (optional).

Returns:

• –

  Dict.

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def read_by_query(self, fields: list = None):
    """Read by Query from Sage Intacct

    Parameters:
        fields (list): Get selective fields to be returned. (optional).

    Returns:
        Dict.
    """
    payload = {
        "readByQuery": {
            "object": self.__dimension,
            "fields": ",".join(fields) if fields else "*",
            "query": None,
            "pagesize": "1000",
        }
    }

    return self.format_and_send_request(payload)

get

get(field, value, fields=None)

Get data from Sage Intacct based on filter.

Parameters:

• field (str) –

  A parameter to filter by the field. (required).

• value (str) –

  A parameter to filter by the field - value. (required).

Returns:

• –

  Dict.

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def get(self, field: str, value: str, fields: list = None):
    """Get data from Sage Intacct based on filter.

    Parameters:
        field (str): A parameter to filter by the field. (required).
        value (str): A parameter to filter by the field - value. (required).

    Returns:
        Dict.
    """
    data = {
        "readByQuery": {
            "object": self.__dimension,
            "fields": ",".join(fields) if fields else "*",
            "query": "{0} = '{1}'".format(field, value),
            "pagesize": "1000",
        }
    }

    return self.format_and_send_request(data)["data"]

get_all

get_all(field=None, value=None, fields=None)

Get all data from Sage Intacct

Returns:

• –

  List of Dict.

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def get_all(self, field: str = None, value: str = None, fields: list = None):
    """Get all data from Sage Intacct

    Returns:
        List of Dict.
    """
    complete_data = []
    count = self.count()
    pagesize = self.__pagesize
    for offset in range(0, count, pagesize):
        data = {
            "query": {
                "object": self.__dimension,
                "select": {
                    "field": fields
                    if fields
                    else dimensions_fields_mapping[self.__dimension]
                },
                "pagesize": pagesize,
                "offset": offset,
            }
        }

        if field and value:
            data["query"]["filter"] = {"equalto": {"field": field, "value": value}}

        paginated_data = self.format_and_send_request(data)["data"][
            self.__dimension
        ]
        complete_data.extend(paginated_data)

    return complete_data

get_by_query

get_by_query(fields=None, and_filter=None, or_filter=None, filter_payload=None)

Get data from Sage Intacct using query method based on filter.

See sage intacct documentation here for query structures: https://developer.intacct.com/web-services/queries/

    Parameters:
        fields (str): A parameter to filter by the field. (required).
        and_filter (list(tuple)): List of tuple containing (operator (str),field (str), value (str))
        or_filter (list(tuple)): List of tuple containing (operator (str),field (str), value (str))
        filter_payload (dict): Formatted query payload in dictionary format.
        if 'between' operators is used on and_filter or or_filter field must be submitted as
        [str,str]
        if 'in' operator is used field may be submitted as [str,str,str,...]

    Returns:
        Dict.

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def get_by_query(
    self,
    fields: List[str] = None,
    and_filter: __query_filter = None,
    or_filter: __query_filter = None,
    filter_payload: dict = None,
):
    """Get data from Sage Intacct using query method based on filter.

    See sage intacct documentation here for query structures:
    https://developer.intacct.com/web-services/queries/

            Parameters:
                fields (str): A parameter to filter by the field. (required).
                and_filter (list(tuple)): List of tuple containing (operator (str),field (str), value (str))
                or_filter (list(tuple)): List of tuple containing (operator (str),field (str), value (str))
                filter_payload (dict): Formatted query payload in dictionary format.
                if 'between' operators is used on and_filter or or_filter field must be submitted as
                [str,str]
                if 'in' operator is used field may be submitted as [str,str,str,...]

            Returns:
                Dict.
    """

    complete_data = []
    count = self.count()
    pagesize = self.__pagesize
    offset = 0
    formatted_filter = filter_payload
    data = {
        "query": {
            "object": self.__dimension,
            "select": {
                "field": fields
                if fields
                else dimensions_fields_mapping[self.__dimension]
            },
            "pagesize": pagesize,
            "offset": offset,
        }
    }
    if and_filter and or_filter:
        formatted_filter = {"and": {}}
        for operator, field, value in and_filter:
            formatted_filter["and"].setdefault(operator, {}).update(
                {"field": field, "value": value}
            )
        formatted_filter["and"]["or"] = {}
        for operator, field, value in or_filter:
            formatted_filter["and"]["or"].setdefault(operator, {}).update(
                {"field": field, "value": value}
            )

    elif and_filter:
        if len(and_filter) > 1:
            formatted_filter = {"and": {}}
            for operator, field, value in and_filter:
                formatted_filter["and"].setdefault(operator, {}).update(
                    {"field": field, "value": value}
                )
        else:
            formatted_filter = {}
            for operator, field, value in and_filter:
                formatted_filter.setdefault(operator, {}).update(
                    {"field": field, "value": value}
                )
    elif or_filter:
        if len(or_filter) > 1:
            formatted_filter = {"or": {}}
            for operator, field, value in or_filter:
                formatted_filter["or"].setdefault(operator, {}).update(
                    {"field": field, "value": value}
                )
        else:
            formatted_filter = {}
            for operator, field, value in or_filter:
                formatted_filter.setdefault(operator, {}).update(
                    {"field": field, "value": value}
                )

    if formatted_filter:
        data["query"]["filter"] = formatted_filter

    for offset in range(0, count, pagesize):
        data["query"]["offset"] = offset
        paginated_data = self.format_and_send_request(data)["data"]
        complete_data.extend(paginated_data[self.__dimension])
        filtered_total = int(paginated_data["@totalcount"])
        if paginated_data["@numremaining"] == "0":
            break
    if filtered_total != len(complete_data):
        warn(
            message="Your data may not be complete. Records returned do not equal total query record count",
            category=DataIntegrityWarning,
            stacklevel=2,
        )
    return complete_data

get_lookup

get_lookup()

Returns all fields with attributes from the object called on.

Source code in hypha/apply/projects/services/sageintacct/wrapper/api_base.py

def get_lookup(self):
    """Returns all fields with attributes from the object called on.

    Parameters:
        self
    Returns:
        Dict.
    """

    data = {"lookup": {"object": self.__dimension}}
    return self.format_and_send_request(data)["data"]