python 使用 Quickbooks Online API v3

python with Quickbooks Online API v3

我需要一些帮助来实现一个可以访问 Quickbooks API 的 python 应用程序。我已经成功编写了几个使用 APIs 的应用程序,但是一旦我们进入 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. 导入错误modules/classes
  2. "glue together" 在 github
  3. 上找到的示例缺少大量代码
  4. 这里不使用 django,我知道上面的请求 class 在 django 中,但我真的很想在不使用 django[=43= 的情况下将其作为 python 脚本工作]
  5. 未从初始 authorize_url 函数中获取 token/identifier/realmId。它打印在屏幕上,但我不确定如何抓取它...

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

* 更新 *

好的,我知道如何让它连接,但我不确定如何获取报告。

答案是这样的,在之前的 API 页面上:

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
        )

仍在尝试获取基本报告...

我对 Python 没有太多经验,但有人与我分享了此代码用于 oauth earlier.If 您对代码还有其他问题,我无法回答。

注意:下面的代码也会调用 V2 QBO api。请不要使用该部分,因为它已被弃用。

看看有没有帮助-

导入Python

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

import xmltodict

class QuickBooks():
    """A wrapper class around Python's Rauth module for 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"

好的,下面是完成这项工作的方法。我专注于报告,下面介绍如何使用 Python:

从 Quickbooks Online API 获取报告

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

2) 确保安装了 rauth。如果您在 AWS/EC2,只需:

sudo yum install rauth

3) 编辑quickbooks2.py文件,在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) 在此处的 Quickbooks 站点上设置沙盒应用程序:https://developer.intuit.com/v2/ui#/app/startcreate(如果您还没有开发者帐户,则必须创建一个)

5) 设置完成后,您可以转到应用程序的 "Keys" 选项卡并获取 App Token、OAuth Consumer Key 和 OAuth Consumer Secret。

6) 前往位于 https://appcenter.intuit.com/Playground/OAuth/IA 的 Intuit Developer Playground 并使用第 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/cQuickbooks Sandbox公司没有Income/Expense2015年的数据,所以你要挑2014年的日期

8) 重要提示:要与 Quickbooks 沙箱一起用于报告目的,您需要更改 get_report() 函数以使用 base_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) 您可以在此处探索一下以测试合适的 URLs:https://developer.intuit.com/apiexplorer?apiname=V3QBO#Reports

12) 报告参考在此处:https://developer.intuit.com/docs/0100_accounting/0400_references/reports,这显示了您可以使用哪些参数。要在资源管理器中测试参数,请在 "Request Body" 部分中输入它们。

我纠结了一段时间,终于弄明白了。希望这对其他人有帮助。