Python for Accessing Finale Inventory REST API

finale_api.py

import requests
import json
import base64
import pprint
import unittest
import time
import datetime

# Put this in contactMechList
ADDRESS_TEMPLATE = {
    "contactMechTypeId": "POSTAL_ADDRESS",
    "infoString": "",
    "address1": "",
    "city": "",
    "stateProvinceGeoId": "",  # 2 char state code
    "postalCode": "",
    "countryGeoId": "USA",
    "address2": "",
    "contactMechPurposeTypeId": "BILLING_LOCATION"  # Choices:"GENERAL_LOCATION", "BILLING_LOCATION","SHIPPING_LOCATION"
}

# Put this in contactMechList
PHONE_NUMBER_TEMPLATE = {
    "contactMechTypeId": "TELECOM_NUMBER",
    "infoString": "the phone number",
    "contactMechPurposeTypeId": "PHONE_WORK"  # Choices include: "PHONE_HOME", "PHONE_WORK",
}                                             # "PHONE_MOBILE", "FAX_NUMBER"

# Put this in contactMechList
EMAIL_ADDRESS = {
    "contactMechTypeId": "EMAIL_ADDRESS",
    "infoString": "[email protected]",
    "contactMechPurposeTypeId": "WORK_EMAIL"  # Choices: "HOME_EMAIL", "WORK_EMAIL", "PAYMENT_EMAIL"
}                                             # "BILLING_EMAIL"


PARTY_GROUP_TEMPLATE = {
    "partyId": None,
    "partyUrl": None,
    "groupName": None,  # this is the person's full name
    "atfLicenseNumber": None,
    "atfLicenseExpiration": None,
    "carrierScac": None,
    "carrierRegistrationNumber": None,
    "carrierRegistrationHazMatNumber": None,
    "hazMatContactTel": None,
    "hazMatContractNumber": None,
    "description": "",
    "roleTypeIdList": ["CUSTOMER"],
    "contactMechList": [],   # Put contact items in here
    "guiOptions": None,
    "userFieldDataList": [{"attrName": "user_10000", "attrValue": ""}],
    "glAccountList": None,
    "statusId": "PARTY_ENABLED",
    "createdDate": None,
    "lastUpdatedDate": None}

# To send party group data, build the template, then json encode it and send.


class Finale(object):
    """
    Manages Finale REST API interactions.

    Either pass url, username and password into __init__ or create a json formatted file called secrets.json
    containing:

    {
        "URL": "https://app.finaleinventory.com/MYCOMPANY/",
        "USERNAME": "MY USERNAME",
        "PASSWORD": "MY PASSWORD"
    }

    __init__ automatically gets an auth cookie and saves it in a requests Session.

    After each REST interaction, the status is stored in self.status_code. With the exception of __init__, an
    unsuccessful status code (e.g. not 200) does not generate an error. So in many cases you will want to
    check it before you proceed.

    If the request is successful, then the results are stored in self.response_data.
    """
    def __init__(self, url=None, username=None, password=None):
        """Login and get save auth cookie in self.session"""
        self.finale_url = u'https://app.finaleinventory.com'
        self.status_code = 200
        self.response_data = None

        if url and username and password:
            self.url = url
        else:
            fp = open(u'secrets.json', 'r')
            secrets = json.load(fp)
            fp.close()
            self.url = secrets[u'URL']
            username = secrets[u'USERNAME']
            password = secrets[u'PASSWORD']

        self.session = requests.Session()
        auth_url = self.url + 'api/auth'
        r = self.session.post(auth_url, data={u'username': username, u'password': password})
        self.status_code = r.status_code
        if r.status_code == 200:
            d = requests.utils.dict_from_cookiejar(self.session.cookies)
            self.session_secret = d[u'JSESSIONID']  # Used for Finale CSRF
        else:
            raise SystemError(u'Error: could not connect to Finale. HTTP status %d' % r.status_code)

    def list_items(self, list_url, **kwargs):
        url = self.url + list_url
        if u'the_filter' in kwargs:
            url += (u'/?filter=%s' % kwargs[u'the_filter'])
        r = self.session.get(url)
        self.status_code = r.status_code

        if r.status_code == 200:
            self.response_data = json.loads(r.text)
        else:
            self.response_data = None

    def get_item(self, item_url):
        r = self.session.get(self.finale_url + item_url)
        self.status_code = r.status_code
        if r.status_code == 200:
            self.response_data = json.loads(r.text)
        else:
            self.response_data = None

    def create_item(self, item_url, data, return_item_key):
        data[u'sessionSecret'] = self.session_secret

        url = self.url + item_url
        json_data = json.dumps(data)
        r = self.session.post(url, data=json_data)
        self.status_code = r.status_code
        if r.status_code == 200:
            self.response_data = json.loads(r.text)
            return self.response_data[return_item_key]
        else:
            self.response_data = None
            return None

    def update_item(self, item_url, data):
        data[u'sessionSecret'] = self.session_secret

        url = self.finale_url + item_url
        json_data = json.dumps(data)

        r = self.session.post(url, data=json_data)
        self.status_code = r.status_code
        if r.status_code == 200:
            self.response_data = json.loads(r.text)
        else:
            self.response_data = None

    def pprint(self):
        """pprints current data."""
        pprint.pprint(self.response_data)

    def list_products(self, **kwargs):
        self.list_items(u'api/product', **kwargs)

    def create_product(self, new_product):
        return self.create_item(u'api/product/', new_product, u'productUrl')

    def list_invoices(self, **kwargs):
        self.list_items(u'api/invoice', **kwargs)

    def list_party_groups(self, **kwargs):
        self.list_items(u'api/partygroup', **kwargs)

    def create_party_group(self, data):
        return self.create_item(u'api/partygroup/', data, u'partyUrl')

    @staticmethod
    def date_filter(start, stop, field_name=u'lastUpdatedDate'):
        """Start and stop are python datetimes. Field names include: lastUpdatedDate """
        s1 = start.replace(microsecond=0).isoformat()
        s2 = stop.replace(microsecond=0).isoformat()
        d = {field_name: [s1, s2]}
        filter_str = base64.urlsafe_b64encode(json.dumps(d))
        return filter_str


#####################################################################################################################
# Tests -------------------------------------------------------------------------------------------------------------
class TestFinale(unittest.TestCase):
    def setUp(self):
        url = u'https://app.finaleinventory.com/demo/'
        username = u'test'
        password = u'finale'
        self.f = Finale(url, username, password)

    def test_products(self):
        print u'Running test_products'
        # See if it works at all
        self.f.list_products()
        self.assertEqual(self.f.status_code, 200)

        # Add a product
        product_number = int(time.time())
        product_id = u"ROI_PID-%d" % product_number
        new_product = {u"productId": product_id, u"internalName": u"ROI test product %d" % product_number}
        new_product_url = self.f.create_product(new_product)
        self.assertEqual(self.f.status_code, 200)

        # Look for new product in list
        self.f.list_products()
        self.assertEqual(self.f.status_code, 200)
        self.assertIn(product_id, self.f.response_data[u'productId'])

        # Update new product
        data = {u'internalName': u"Updated ROI test product %d" % product_number}
        self.f.update_item(new_product_url, data)
        self.assertEqual(self.f.status_code, 200)

    def test_party_groups(self):
        print u'Running test_party_groups'
        # See if it works at all
        self.f.list_party_groups()
        self.assertEqual(self.f.status_code, 200)

        # Add a customer (party_group) - data fields grabbed from browser
        party_number = int(time.time())
        group_name = "Test %d" % party_number
        data = {"partyId": None,
                "partyUrl": None,
                "groupName": group_name,
                "atfLicenseNumber": None,
                "atfLicenseExpiration": None,
                "carrierScac": None,
                "carrierRegistrationNumber": None,
                "carrierRegistrationHazMatNumber": None,
                "hazMatContactTel": None,
                "hazMatContractNumber": None,
                "description": "Some notes",
                "roleTypeIdList": ["CUSTOMER"],
                "contactMechList": [{"contactMechTypeId": "POSTAL_ADDRESS",
                                     "infoString": "",
                                     "address1": "2681 Norwich",
                                     "city": "Fitchburg",
                                     "stateProvinceGeoId": "WI",
                                     "postalCode": "53711",
                                     "countryGeoId": "USA",
                                     "address2": "addition",
                                     "contactMechPurposeTypeId": "BILLING_LOCATION"},

                                    {"contactMechTypeId": "TELECOM_NUMBER",
                                     "infoString": "6087778888",
                                     "contactMechPurposeTypeId": "PHONE_WORK"},

                                    {"contactMechTypeId": "EMAIL_ADDRESS",
                                     "infoString": "[email protected]",
                                     "contactMechPurposeTypeId": "WORK_EMAIL"}],
                "guiOptions": None,
                "userFieldDataList": [{"attrName": "user_10000", "attrValue": ""}],
                "glAccountList": None,
                "statusId": "PARTY_ENABLED",
                "createdDate": None,
                "lastUpdatedDate": None}

        new_party_group_url = self.f.create_party_group(data)
        self.assertEqual(self.f.status_code, 200)
        #print self.f.response_data

        # Look for new party group in list
        self.f.list_party_groups()
        self.assertEqual(self.f.status_code, 200)
        self.assertIn(new_party_group_url, self.f.response_data[u'partyUrl'])

        # Update new party group
        data = {u'description': u"Updated description"}
        self.f.update_item(new_party_group_url, data)
        self.assertEqual(self.f.status_code, 200)

    def test_filter(self):
        print u'Test filtering'
        # See if it works at all
        self.f.list_party_groups()
        self.assertEqual(self.f.status_code, 200)
        unfiltered_length = len(self.f.response_data[u'partyId'])

        # Run again with a filter
        stop = datetime.datetime.now()
        start = stop - datetime.timedelta(days=2)
        fs = self.f.date_filter(start, stop)
        self.f.list_party_groups(the_filter=fs)
        self.assertEqual(self.f.status_code, 200)
        filtered_length = len(self.f.response_data[u'partyId'])

        print 'N Unfiltered: ', unfiltered_length
        print 'N Filtered: ', filtered_length
        self.assertGreater(unfiltered_length, filtered_length)


if __name__ == '__main__':
    unittest.main()


"""
The MIT License (MIT)

Copyright (c) 2104 Charles Martin

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
"""