【问题标题】:Xero : OAuth2: Python3: Example Code to get the refresh tokenXero:OAuth2:Python3:获取刷新令牌的示例代码
【发布时间】:2020-12-10 09:44:49
【问题描述】:

Xero 已将其 API 更改为需要 OAuth2 连接而不是 OAuth1。

我在 OAuth1 中有一个可行的解决方案,但 Oauth2 的示例充其量很少,但主要用于网站。

我是另一位成功创建 Oauth1 解决方案的开发人员,该解决方案成功地用作机器对机器的解决方案,不涉及 Web 服务器。

Xero 有一些在 Postman 中运行的示例,可以在我的测试环境中轻松运行。

我正在尝试重现 Python3 中刷新令牌的邮递员操作。

下面的基本代码是我目前所在的位置:

#client_id = "xxxxx"
#client_secret = "xxxxx"
callback_url = "https://api.xero.com/connections"
re_directURI = "https://developer.xero.com"
scopes = "offline_access accounting.contacts accounting.transactions"

refresh_url = "https://identity.xero.com/connect/token"

access_token = open('AccessToken.txt').read()
old_refresh_token = open('RefreshToken.txt','r').read()


# Refresh Token code...

import requests

#def refresh_xero_token(refresh_token):

headers = {
'grant_type': 'refresh_token',
'Content-Type': 'application/json',
}
data = {
'grant_type': 'refresh_token',
'refresh_token': old_refresh_token,
'client_id': client_id,
'client_secret': client_secret
}
print(data,headers)
response = requests.post(refresh_url, headers=headers, data=data)

#return response.json()
print(response.text)

到目前为止,我还没有找到一个可以在没有 Web 服务器的情况下工作的示例,只是使用 python 与 Xero 服务器通信以将本地数据传输到零 API。

使用 xoauth,.exe (windows) 获取 access_token,然后在 postman 中我可以运行示例以刷新令牌、连接、发票等到演示公司。

而且我相信能够复制这些示例将为我提供获得有效解决方案所需的一切。

目前我只得到这个 python 代码 {“错误”:“无效请求”}

所以,显然我错过了一些东西。

我将自己归类为 Python 或 Oauth2 的新手,但由于我之前使用 Oauth1 连接的解决方案取得了成功,因此选择了这条路径。

我会询问 Xero 开发者社区,但我写这篇文章是为了让我们软件的用户将数据推送到他们的 Xero 帐户中,因此为了进行测试,我只有一个试用帐户,我无法访问Xero 开发者社区。​​p>

这本身就很烦人。

Xero 支持似乎也没什么用,我试过了。

如果有人可以提供帮助,那就太好了。

提前感谢您提供的任何帮助。

【问题讨论】:

    标签: python api oauth-2.0 xero


    【解决方案1】:

    在使用 xoauth 应用程序并设置连接后,我发现使用刷新令牌,运行此功能可以保持连接正常运行。

    def refresh_xero_token():
        refresh_url =       "https://identity.xero.com/connect/token"
    
        old_refresh_token = open('RefreshToken.txt','r').read()
    
        tokenb4 = f"{client_id}:{client_secret}"
        basic_token = base64.urlsafe_b64encode(tokenb4.encode()).decode()
    
        headers = {
          'Authorization': f"Basic {basic_token}",
          'Content-Type': 'application/x-www-form-urlencoded',
        }
        data = {
        'grant_type': 'refresh_token',
        'refresh_token': old_refresh_token
        }
    
        try:
            response = requests.post(refresh_url, headers=headers, data=data)
    
            results = response.json()
            open('RefreshToken.txt','w').write(results["refresh_token"])
            open('AccessToken.txt','w').write(results["access_token"])
    
        except Exception as err:
            print("ERROR ! Refreshing token error?")
            print(response.text)
    

    【讨论】:

      【解决方案2】:

      作为附加信息,我还可以使用此连接,例如在 Xero 中创建联系人:

      在此示例中,irContact 是 SQL 表中的 SQLAlchemy 数据行。

      def create_contact(connection, irContact, access_token):

      #Setup new contact
      address1 = {"AddressType": "POBOX"}
      
      if irContact['addressline1'] is not None: address1.update({"AddressLine1": irContact['addressline1']})
      if irContact['addressline2'] is not None: address1.update({"AddressLine2": irContact['addressline2']})
      if irContact['addressline3'] is not None: address1.update({"AddressLine3": irContact['addressline3']})
      if irContact['addressline4'] is not None: address1.update({"AddressLine4": irContact['addressline4']})
      if irContact['city'] is not None: address1.update({"City": irContact['city']})
      if irContact['region'] is not None: address1.update({"Region": irContact['region']})
      if irContact['postalcode'] is not None: address1.update({"PostalCode": irContact['postalcode']})
      if irContact['country'] is not None: address1.update({"Country": irContact['country']})
      if irContact['attentionto'] is not None: address1.update({"AttentionTo": irContact['attentionto']})
      
      #print (address1.values())
      
      addresses = []
      addresses.append(address1)
      
      phones = []
      if irContact['phonenumber'] is not None:
          phone1 = {"PhoneType": "DEFAULT"}
          #phone1.update({"PhoneType": "DEFAULT"})
          if irContact['phonenumber'] is not None: phone1.update({"PhoneNumber": irContact['phonenumber']})
          if irContact['phoneareacode'] is not None: phone1.update({"PhoneAreaCode": irContact['phoneareacode']})
          if irContact['phonecountrycode'] is not None: phone1.update({"PhoneCountryCode": irContact['phonecountrycode']})
      
          phones.append(phone1)
          #print (phone1.values())
          
      
      if irContact['mobilenumber'] is not None:
          phone2 = {"PhoneType": "MOBILE"}
          if irContact['phonenumber'] is not None: phone2.update({"PhoneNumber": irContact['mobilenumber']})
          if irContact['phoneareacode'] is not None: phone2.update({"PhoneAreaCode": irContact['mobileareacode']})
          if irContact['phonecountrycode'] is not None: phone2.update({"PhoneCountryCode": irContact['mobilecountrycode']})
      
          phones.append(phone2)
          #print (phone2.values())
          
      
      contact = { "Name": irContact['name'],
                   "ContactNumber": irContact['contactnumber'],
                   "AccountNumber": irContact['accountnumber'],
                   #"ContactStatus": "ACTIVE",
                   "FirstName": irContact['firstname'],
                   "LastName": irContact['lastname'],
                   #"EmailAddress": irContact['emailaddress'],
                   "Addresses": addresses,
                   #"Phones":phones
                  }
      
      contacts = [contact]
      #print(contacts)
      
      contacts_url = "https://api.xero.com/api.xro/2.0/Contacts"
      
      headers = {
        'Authorization': f"Bearer {access_token}",
        'Accept': 'application/json',
        'Content-Type': 'application/json',
        'xero-tenant-id': tenant_id,
      }
      data = {
      'Contacts': [contact],
      }
      #print(data)
      
      try:
          response = requests.post(contacts_url, headers=headers, json=data)
      except Exception as err:
          print("ERROR! Contact: %s" % (str(err) ))
          print(response.text)
          return 0
      
      #print(response.text)
      
      results = response.json()
      
      if 'Contacts' in results:
          newcontacts = results['Contacts']
          for newcontact in newcontacts: break
      
          query = "update xero_contact set errortext='', ContactID='%s' where id=%d" % (newcontact["ContactID"], irContact['id'])
          connection.execute(query)
      
          query = "update xero_invoice_header set ContactID='%s' where OurContactID=%d and (InvoiceID='' or InvoiceID is null ) " % ( newcontact["ContactID"], irContact['id']  )
          connection.execute(query)
      

      我相信,有了这么多信息,任何人都可以创建自己的 Xero 机器对机器接口...... 意识到可以通过对请求调用的标头和/或数据元素的最小调整来读取和创建其他记录。

      我发现缺少此信息非常令人沮丧,如果人们能找到此信息,将来可能会对他们有所帮助。

      【讨论】:

        猜你喜欢
        • 1970-01-01
        • 2012-05-04
        • 2012-06-29
        • 1970-01-01
        • 2019-10-26
        • 2015-11-28
        • 2020-09-25
        • 2017-05-01
        • 2018-06-23
        相关资源
        最近更新 更多