Python with Quickbooks Online v3 API

I need help implementing a python application that accesses the Quickbooks API. I have successfully written several applications that use the API, but once we get into the OAuth world, I get a little lost.

Anyway, I found the cover of quickbooks-python: https://github.com/troolee/quickbooks-python

but there are zero working code examples showing how to properly implement. I am guessing that a more experienced python programmer could figure out how to make this work without any instructions, but it seems to me that I am missing a reason.

If I could connect to it, I could make it work from there ...

It seems like the documentation about github jumping around and for a more experienced programmer would probably make sense. But I just don't follow ...

from quickbooks import *

consumerKey =           "fromApiConsole"
consumerSecret =        "fromApiConsole"
callbackUrl =           "https://quickbooks.api.intuit.com/v3"

qbObject = QuickBooks(
        consumer_key = consumerKey,
        consumer_secret = consumerSecret,
        callback_url = callbackUrl
        )

authorize_url = qbObject.get_authorize_url() # will create a service, and further set up the qbObject.

oauth_token = request.GET['oauth_token']
oauth_verifier = request.GET['oauth_verifier']
realm_id = request.GET['realmId']

session = qbObject.get_access_tokens(oauth_verifier)

# say you want access to the reports

reportType = "ProfitAndLoss"

url = "https://quickbooks.api.intuit.com/v3/company/asdfasdfas/"
url += "reports/%s" % reportType

r = session.request( #This is just a Rauth request
    "POST",
    url,
    header_auth = True,
    realm = realm_id,
    params={"format":"json"}
    )

qb = QuickBooks(
    consumer_key = consumerKey,
    consumer_secret = consumerSecret,
    access_token = qbtoken.access_token, # the stored token
    access_token_secret = qbtoken.access_token_secret, # the stored secret
    company_id = qbtoken.realm_id #the stored realm_id
    )

qbText = str(qb.query_objects(business_object, params, query_tail))

print qbText

      

I am sure that I:

  • import of wrong modules / classes
  • missing huge pieces of code to "glue" the samples found in github
  • not using django here and I know the request class above is in django, but I'd really like to just make this work as a python script without using django
  • don't get token / id / realmId from initial authorize_url function. it prints to the screen, but I'm not sure how to capture it ...

The end goal here is to simply connect and get a P&L operator from Quickbooks Online. If I can get this far, I'm sure I can get the rest of what I need from the API. I don't really need to CHANGE everything, I just want to include data from reports in some dashboards.

* UPDATE *

OK, I figured out how to connect it, but I'm not sure how to get to the reports.

the answer was like the one on the previous API page:

Accessing the API
Once you've gotten a hold of your QuickBooks access tokens, you can create a QB object:

qb = QuickBooks(consumer_key = QB_OAUTH_CONSUMER_KEY, 
        consumer_secret = QB_OAUTH_CONSUMER_SECRET,
        access_token = QB_ACCESS_TOKEN, 
        access_token_secret = QB_ACCESS_TOKEN_SECRET,
        company_id = QB_REALM_ID
        )

      

still trying to get basic reports ...

+1


source to share


2 answers


Ok, so here's how to do it. I am focused on reports, so here's how you can get reports from the Quickbooks Online API using Python:

1) Go to https://github.com/finoptimal-dev/quickbooks-python and download the code

2) Make sure you have a walkie-talkie installed. If you are on AWS / EC2, simply:

sudo yum install rauth

      

3) Edit the quickbooks2.py file and add the following to the END:

qb = QuickBooks(consumer_key = QB_OAUTH_CONSUMER_KEY, 
        consumer_secret = QB_OAUTH_CONSUMER_SECRET,
        access_token = QB_ACCESS_TOKEN, 
        access_token_secret = QB_ACCESS_TOKEN_SECRET,
        company_id = QB_REALM_ID
        )

      

4) Set up your sandbox app on the Quickbooks site here: https://developer.intuit.com/v2/ui#/app/startcreate (you will need to create a developer account, t already there)

5) Once configured, you can go to the Application Keys tab and grab the Application Token, OAuth Consumer Key and OAuth User Secret.

6) Go to the Intuit Developer Playground at https://appcenter.intuit.com/Playground/OAuth/IA and use the information from step # 5 to get the Access Token and Secret Access Token.

7) Change the variables in step 3 to the correct values. For QB_REALM_ID, this is the company ID. You can get this in the sandbox by going into https://developer.intuit.com/v2/ui#/sandbox and looking for the company ID.

7) add the following code below code from step # 3 above

print qb.get_report('ProfitAndLoss','summarize_column_by=Month&start_date=2014-01-01&end_date=2014-12-31')

      

I am using the above dates. b / c Quickbooks Sandbox has no revenue / expense data for 2015, so you need to pick dates in 2014.



8) IMPORTANT: For use with Quickbox Sandbox for reporting purposes, you need to modify the get_report () function to use base_url_v3 instead of hardcoding for the production URL.

Look for a line in the get_report () function that looks like this:

url = "https://quickbooks.api.intuit.com/v3/company/%s/" % \

      

and change it to this:

url = self.base_url_v3 + "/company/%s/" % \

      

9) Now you can completely change base_url_v3 from above:

base_url_v3 =  "https://sandbox-quickbooks.api.intuit.com/v3"

      

10) You should now be working:

python quickbooks2.py

      

You should see a bunch of JSON data from Quickbooks Sandbox.

11) You can explore the following urls a bit: https://developer.intuit.com/apiexplorer?apiname=V3QBO#Reports

12) The link to the report is here: https://developer.intuit.com/docs/0100_accounting/0400_references/reports and this shows you what options you can use. To check the parameters in Explorer, you enter them in the Request Body section.

I have been struggling with this for a while and finally figured it out. Hope this helps someone else.

+2


source


I don't have much experience with Python, but someone shared this code with me for oauth earlier. If you have additional questions about the code, I cannot answer them.

NOTE. The code below also calls the V2 QBO apis. Do not use this part as it is outdated.

See if this helps -

Python import



from import routine OAuth1Session, OAuth1Service import xml.etree.ElementTree as ET

import xmltodict

class QuickBooks (): "" Wrapper class around the Quickbooks API Python Rauth module "" "

access_token = ''
access_token_secret = ''
consumer_key = ''
consumer_secret = ''
company_id = 0
callback_url = ''
session = None

base_url_v3 =  "https://quickbooks.api.intuit.com/v3"
base_url_v2 = "https://qbo.intuit.com/qbo1"

request_token_url = "https://oauth.intuit.com/oauth/v1/get_request_token"
access_token_url = "https://oauth.intuit.com/oauth/v1/get_access_token"

authorize_url = "https://appcenter.intuit.com/Connect/Begin"

# Things needed for authentication
qbService = None

request_token = ''
request_token_secret = ''


def __init__(self, **args):

    if 'consumer_key' in args:
        self.consumer_key = args['consumer_key']

    if 'consumer_secret' in args:
        self.consumer_secret = args['consumer_secret']

    if 'access_token' in args:
        self.access_token = args['access_token']

    if 'access_token_secret' in args:
        self.access_token_secret = args['access_token_secret']

    if 'company_id' in args:
        self.company_id = args['company_id']

    if 'callback_url' in args:
        self.callback_url = args['callback_url']


def get_authorize_url(self):
    """Returns the Authorize URL as returned by QB, 
    and specified by OAuth 1.0a.
    :return URI:
    """
    self.qbService = OAuth1Service(
            name = None,
            consumer_key = self.consumer_key,
            consumer_secret = self.consumer_secret,
            request_token_url = self.request_token_url,
            access_token_url = self.access_token_url,
            authorize_url = self.authorize_url,
            base_url = None
        )
    self.request_token, self.request_token_secret = self.qbService.get_request_token(
            params={'oauth_callback':self.callback_url}
        )

    return self.qbService.get_authorize_url(self.request_token)




def get_access_tokens(self, oauth_verifier):
    """Wrapper around get_auth_session, returns session, and sets 
    access_token and access_token_secret on the QB Object.
    :param oauth_verifier: the oauth_verifier as specified by OAuth 1.0a
    """
    session = self.qbService.get_auth_session(
            self.request_token, 
            self.request_token_secret,
            data={'oauth_verifier': oauth_verifier})

    self.access_token = session.access_token
    self.access_token_secret = session.access_token_secret

    return session



def create_session(self):
    if self.consumer_secret and self.consumer_key and self.access_token_secret and self.access_token:
        # print "hi"
        session = OAuth1Session(self.consumer_key,
            self.consumer_secret,
            self.access_token,
            self.access_token_secret,
            )
        # print session
        self.session = session
    else:
        pass
        #TODO: raise an error
    return self.session


def keep_trying(self, r_type, url, header_auth, realm, payload=''):

    if self.session != None:
        session = self.session
    else:
        session = self.create_session()
        self.session = session

    trying = True
    tries = 0
    while trying:
        print url
        tries += 1
        if "v2" in url:
            r = session.request(r_type, url, header_auth, realm, data=payload)

            r_dict = xmltodict.parse(r.text)
            # print "DICT", r_dict
            if "FaultInfo" not in r_dict or tries > 4:
                trying = False
        else:
            # url = "https://qb.sbfinance.intuit.com/v3/company/184010684/query?query=SELECT * FROM JournalEntry"
            # url = "https://quickbooks.api.intuit.com/v3/company/184010684/journalentry/24772"
            # url = "https://quickbooks.api.intuit.com/v3/company/184010684/query?query='SELECT+*+FROM+JournalEntry'"
            # url = "https://qb.sbfinance.intuit.com/v3/company/184010684/query?query=SELECT%20%2A%20FROM%20JournalEntry&"
            print url, r_type
            headers = {'Accept': 'application/json'}
            r = session.request(r_type, url, header_auth, realm, headers = headers)
            # r.headers
            print "\n\n INITIAL TEXT \n\n", r.text


            print "request headers:", r.request.headers
            print "request URL:", r.request.url
            print "response headers:", r.headers

            r_dict = r.text
            if "Fault" not in r_dict or tries > 4:
                trying = False
            r_dict = []

    return r_dict

def fetch_customer(self, pk):

    if pk:
        url = self.base_url_v2 + "/resource/customer/v2/%s/%s" % ( self.company_id, pk)
        r_dict = self.keep_trying("GET", url, True, self.company_id)

        return r_dict['Customer']


def fetch_customers(self, all=False, page_num=0, limit=10):
    if self.session != None:
        session = self.session
    else:
        session = self.create_session()
        self.session = session

    # We use v2 of the API, because what the fuck, v3.
    url = self.base_url_v2
    url += "/resource/customers/v2/%s" % (self.company_id)

    customers = []

    if all:
        counter = 1
        more = True

        while more:
            payload = {
                "ResultsPerPage":30,
                "PageNum":counter,
                }

            trying = True

            # Because the QB API is so iffy, let try until we get an non-error

            # Rewrite this to use same code as above.
            while trying:
                r = session.request("POST", url, header_auth = True, data = payload, realm = self.company_id)
                root = ET.fromstring(r.text)
                if root[1].tag != "{http://www.intuit.com/sb/cdm/baseexceptionmodel/xsd}ErrorCode":
                    trying = False
                else:
                    print "Failed"

            session.close()
            qb_name = "{http://www.intuit.com/sb/cdm/v2}"

            for child in root:
                # print child.tag, child.text
                if child.tag == "{http://www.intuit.com/sb/cdm/qbo}Count":

                    if int(child.text) < 30:
                        more = False
                        print "Found all customers"

                if child.tag == "{http://www.intuit.com/sb/cdm/qbo}CdmCollections":
                    for customer in child:

                        customers += [xmltodict.parse(ET.tostring(customer))]

            counter += 1

            # more = False
            # print more

    else:

        payload = {
            "ResultsPerPage":str(limit),
            "PageNum":str(page_num),
            }

        r = session.request("POST", url, header_auth = True, data = payload, realm = self.company_id)

        root = ET.fromstring(r.text)

        #TODO: parse for all customers


    return customers

def fetch_sales_term(self, pk):
    if pk:
        url = self.base_url_v2 + "/resource/sales-term/v2/%s/%s" % ( self.company_id, pk)
        r_dict = self.keep_trying("GET", url, True, self.company_id)
        return r_dict


def fetch_invoices(self, **args):
    if "query" in args:
        payload = ""
        if "customer" in args['query']:
            payload = {
                "Filter":"CustomerId :Equals: %s" % (args['query']['customer'])
            }


        # while more:
        url = self.base_url_v2 + "/resource/invoices/v2/%s/" % (self.company_id)
        r_dict = self.keep_trying("POST", url, True, self.company_id, payload)
        invoices = r_dict['qbo:SearchResults']['qbo:CdmCollections']['Invoice']

        return invoices
    elif "pk" in args:
        # TODO: Not tested
        url = self.base_url_v2 + "/resource/invoice/v2/%s/%s" % ( self.company_id, args['pk'])
        r_dict = self.keep_trying("GET", url, True, self.company_id)
        return r_dict
    else:
        url = self.base_url_v2 + "/resource/invoices/v2/%s/" % (self.company_id)
        r_dict = self.keep_trying("POST", url, True, self.company_id, payload)
        return "BLAH"


def fetch_journal_entries(self, **args):
    """ Because of the beautiful way that journal entries are organized
    with QB, you're still going to have to filter these results for the
    actual entity you're interested in. Luckily it only returns the entries
    that are relevant to your search

    :param query: a dictionary that includes 'customer', and the QB id of the
        customer
    """

    if "query" in args:
        payload = {}
        more = True
        counter = 1
        journal_entries = []

        if "customer" in args['query']:

            payload = {
                "Filter":"CustomerId :Equals: %s" % (args['query']['customer'])
            }

            # payload = {
            #     "query":"SELECT * FROM JournalEntry",
            # }

        while more:

            payload["ResultsPerPage"] = 30

            payload["PageNum"] = counter

            # url = self.base_url_v2 + "/resource/journal-entries/v2/%s/" % (self.company_id)
            # url = self.base_url_v3 + "/company/%s/query" % (self.company_id)
            url = "https://qb.sbfinance.intuit.com/v3/company/184010684/query?query=SELECT%20%2A%20FROM%20JournalEntry&"

            r_dict = self.keep_trying("GET", url, True, self.company_id, payload)

            more = False
            # print r_dict['qbo:SearchResults']['qbo:Count']
            counter = counter + 1
            # if int(r_dict['qbo:SearchResults']['qbo:Count']) < 30:
                # more = False

            # journal_entry_set = r_dict['qbo:SearchResults']['qbo:CdmCollections']['JournalEntry']
            # journal_entries += [journal_entry_set]
        return []
        # return r_dict['qbo:SearchResults']['qbo:CdmCollections']['JournalEntry']

    elif "pk" in args:
        # TODO: Not Tested
        url = self.base_url_v2 + "/resource/journal-entry/v2/%s/%s" % ( self.company_id, args['pk'])
        r_dict = self.keep_trying("GET", url, True, self.company_id)
        return r_dict
    else:
        url = self.base_url_v2 + "/resource/journal-entries/v2/%s/" % (self.company_id)
        r_dict = self.keep_trying("POST", url, True, self.company_id)
        print r_dict
        return "BLAH"

      

0


source







All Articles