python与Quickbooks Online API v3

Question

我需要一些帮助来实现一个访问Quickbooks API的python应用程序。我已经成功地编写了几个使用api的应用程序，但是一旦我们进入OAuth世界，我就有点不知所措了。在

无论如何，我在这里找到了quickbooks python包装： https://github.com/troolee/quickbooks-python

但是，没有任何工作代码的例子说明如何正确实现。我想一个更有经验的python程序员可以在没有任何指令的情况下解决这个问题，但是我似乎缺少了一些基本知识。在

如果我能把它连接起来，我也许能让它在那里工作。。。在

似乎github上的文档跳转了一下，对于一个更有经验的程序员来说，可能是非常有意义的。但我只是不明白。。。在

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

我确信我：

1. 导入错误的模块/类
2. 缺少大量代码来“粘合”github上找到的样本
3. 这里不使用django，而且我知道上面的请求类是用django编写的，但是我真的想让它作为一个python脚本而不使用django
4. 无法从初始authorize\u url函数获取令牌/标识符/realmId。它会在屏幕上打印出来，但我不知道如何抓住它。。。在

这里的最终目标实际上只是连接并从Quickbooks Online获取损益表。如果我能做到这一点，我相信我可以从API中得到我所需要的其余部分。我真的不需要更改任何内容，我只是希望将报表中的数据包含到一些仪表板中。在

*更新*

好吧，我想好了怎么把它连接起来，但我不知道怎样才能得到报告。在

答案是这样的，在之前的API页面上：

^{pr2}$

还在努力得到基本报告。。。在

Answer 1

但是我没有和oauth分享过很多关于Python的经验早一点。如果我不能回答你的附加问题。在

注意：下面的代码还调用V2 QBO api。请不要使用该零件，因为它已被弃用。

看看有没有帮助-

导入Python

从rauth导入OAuth1Session，OAuth1Service 进口xml.etree.ElementTree作为ET

导入xmltodict

类QuickBooks（）： “”“一个围绕Python Rauth模块的包装类，用于Quickbooks the API”“”

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's 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"

Answer 2

好吧，下面是如何让这个工作。我的重点是报告，下面是如何使用Python从Quickbooks Online API获取报告：

1）转到https://github.com/finoptimal-dev/quickbooks-python并下载代码

2）确保安装了rauth。如果您使用的是AWS/EC2，只需：

sudo yum install rauth

2.bookspy添加以下bookspy文件：

^{pr2}$

4）在Quickbooks站点上设置沙盒应用程序：https://developer.intuit.com/v2/ui#/app/startcreate（如果您还没有开发人员帐户，则必须创建一个开发人员帐户）

5）安装完成后，您可以转到应用程序的“密钥”选项卡并获取应用程序令牌、OAuth消费者密钥和OAuth消费者机密。在

6）去Intuit开发者游乐场https://appcenter.intuit.com/Playground/OAuth/IA，使用步骤5中的信息获取访问令牌和访问令牌机密。在

7）将步骤3中的变量更改为正确的值。对于QB_REALM_ID，这是公司ID。您可以通过登录https://developer.intuit.com/v2/ui#/sandbox并查找公司ID在沙盒中获取

7）在上述步骤3的代码下方添加以下代码

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

我使用上面的日期b/c Quickbooks Sandbox公司在2015年没有收入/支出数据，所以你必须选择2014年的日期。在

8）重要提示：要将Quickbooks沙盒用于报告目的，您需要更改get_report（）函数，以使用基本的\u url_v3，而不是硬编码为生产url。在

在get_report（）函数中查找如下所示的行：

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

把它改成：

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

9）现在您可以在顶部将base_url_v3更改为：

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

10）现在您应该能够运行：

python quickbooks2.py

您应该会看到来自Quickbooks Sandbox公司的一堆JSON数据。在

11）您可以探索一下，在这里测试合适的url:https://developer.intuit.com/apiexplorer?apiname=V3QBO#Reports

12）报告引用在这里：https://developer.intuit.com/docs/0100_accounting/0400_references/reports，它显示了可以使用哪些参数。要在资源管理器中测试参数，请在“请求正文”部分输入参数。在

我为此挣扎了一段时间，终于弄明白了。希望这对其他人有帮助。在