python with 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. 沒有得到令牌/ IDE來自初始authorize_url函數的ntifier/realmId。它打印在屏幕上，但我不知道如何抓住它...

這裏的最終目標只是連接並從Quickbooks Online獲取P & L聲明。如果我可以得到那麼多，我確信我可以從API中獲得所需的其餘部分。我並不需要改變任何東西，我只是想將報告中的數據包含在某些儀表板中。

* UPDATE *

好吧，我想通了，怎麼把它連接，但我不知道怎麼去報告。

的答案是這樣的，這是之前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 
     ) 

仍試圖獲得基本的報告...

Answer 1

好了，所以這裏是如何使 「周圍的Python的Rauth模塊的Quickbooks的API包裝類」這項工作。我專注於報告，所以這裏是你如何可以使用Python得到QuickBooks的在線API報告：

1）進入​​3210和下載代碼

2）確保你已經安裝了rauth。如果您在AWS/EC2，簡單地說：

sudo yum install rauth 

3）編輯quickbooks2.py文件，並添加以下到最後：

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）設置完成後，你可以進入App的「Keys」選項卡並獲取App Token，OAuth Consumer密鑰和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') 

我使用上述日期二/三的Quickbooks沙箱公司沒有收入/支出數據在2015年，所以得挑選日期於2014年。

8）重要提示：要使用Quickbooks Sandbox進行報告，您需要更改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）您可以探索了一下，在這裏考出相應的網址：https://developer.intuit.com/apiexplorer?apiname=V3QBO#Reports

12）本報告引用是在這裏：，這顯示了哪些參數可以使用。要在資源管理器中測試參數，請在「請求正文」部分輸入它們。

我掙扎了一會兒，終於弄明白了。希望這可以幫助別人。

Answer 2

我沒有與Python多少經驗，但有人共享這個代碼與我的oauth早些時候。如果你有關於代碼的其他問題，我將無法回答他們。

注意：下面的代碼還會調用V2 QBO apis。請不要使用該部分，因爲它已過時。

看它是否helps-

導入的Python

從rauth進口OAuth1Session，OAuth1Service 進口xml.etree。ElementTree，一個ET

進口xmltodict

類的QuickBooks（）： 「」 「」

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"