Xero Python拉取会计数据OAuth2报错 无需Flask的自动化方案咨询

报错原因 #

xero-python SDK 要求你为 ApiClient 配置 oauth2_token_getter 回调函数，用来告诉 SDK 从哪里获取当前有效的 access token，你初始化配置时未提供该函数，因此触发该报错。

修复方案 #

你已经自己实现了令牌刷新逻辑，只需要在初始化 ApiClient 后，注册 token 获取回调即可，不需要引入任何 Web 框架。

在你初始化 api_client 的代码之后，添加如下注册逻辑：

# 注册token获取回调，SDK调用接口时会自动调用该函数拿access token
@api_client.oauth2_token_getter
def get_valid_access_token():
    with open(acces_token_dir, 'r', encoding='utf-8') as f:
        return f.read().strip()

修改完后即可正常调用接口拉取数据，以下是你原代码修复后的完整可运行版本（已替换转义字符）：

import dateutil
import os
import requests
import base64
from xero_python.accounting import AccountingApi
from xero_python.api_client import ApiClient
from xero_python.api_client.configuration import Configuration
from xero_python.api_client.oauth2 import OAuth2Token
from xero_python.exceptions import AccountingBadRequestException

basedir = os.path.abspath(os.path.dirname(__file__))
dst_dir=basedir + f'/utils/data/refresh_token.txt'
tenant_dir=basedir + f'/utils/data/tenantId.txt'
acces_token_dir=basedir + f'/utils/data/accesstokenId.txt'

client_id = 'xxxxxx'
client_secret = 'xxxxxxx'
scope = 'offline_access accounting.journals.read accounting.settings.read accounting.transactions.read'
# 提前生成base64编码的client_id:client_secret，避免每次刷新重复计算
b64_id_secret = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()

def XeroRefreshToken(refresh_token):
    token_refresh_url = 'https://identity.xero.com/connect/token'
    response = requests.post(token_refresh_url,
                             headers={
                                 'Authorization': 'Basic ' + b64_id_secret,
                                 'Content-Type': 'application/x-www-form-urlencoded'
                             },
                             data={
                                 'grant_type': 'refresh_token',
                                 'refresh_token': refresh_token
                             })
    response.raise_for_status()
    json_response = response.json()

    new_refresh_token = json_response['refresh_token']
    with open(dst_dir, 'w', encoding='utf-8') as f:
        f.write(new_refresh_token)

    new_access_token = json_response['access_token']
    with open(acces_token_dir, 'w', encoding='utf-8') as f:
        f.write(new_access_token)
    return new_access_token, new_refresh_token

def XeroRequests():
    old_refresh_token = open(dst_dir,'r').read()
    new_token = XeroRefreshToken(old_refresh_token)
    # 注意：你原来的XeroTenants函数自行实现，用于获取租户ID
    xero_tenant_id = XeroTenants(new_token[0])
    with open(tenant_dir, 'w', encoding='utf-8') as f:
        f.write(xero_tenant_id)
    return xero_tenant_id

# 先刷新令牌，保证本地存储的是最新的有效令牌
XeroRequests()
xero_tenant_id = open(tenant_dir,'r').read()

# 初始化ApiClient
api_client = ApiClient(
    Configuration(
        debug=False,
        oauth2_token=OAuth2Token(
            client_id=client_id, client_secret=client_secret
        ),
    ),
    pool_threads=1,
)

# 注册token获取回调
@api_client.oauth2_token_getter
def get_valid_access_token():
    with open(acces_token_dir, 'r', encoding='utf-8') as f:
        return f.read().strip()

# 拉取会计科目示例
def accounting_get_accounts(xero_tenant_id):
    api_instance = AccountingApi(api_client)
    if_modified_since = dateutil.parser.parse("2020-02-06T12:17:43.202-08:00")
    where = 'Status=="ACTIVE" AND Type=="BANK"'
    order = 'Name ASC'

    try:
        api_response = api_instance.get_accounts(xero_tenant_id, if_modified_since=if_modified_since, where=where, order=order)
        print(api_response)
    except AccountingBadRequestException as e:
        print("Exception when calling AccountingApi->getAccounts: %s\n" % e)

accounting_get_accounts(xero_tenant_id)

权限适配 #

你当前的scope仅支持读取日记账，若需要拉取发票、会计科目表，需要先到Xero开发者平台修改应用的权限配置，添加对应scope：

• 会计科目表：accounting.settings.read
• 发票：accounting.transactions.read
  修改后需要重新走一次授权流程，获取新的refresh token替换你本地存储的旧令牌。

定时自动执行方案 #

不需要引入Web框架，直接用系统定时任务即可实现每日夜间自动执行：

• Linux/Mac系统：配置crontab，例如每天凌晨2点执行脚本：

  0 2 * * * /usr/bin/python3 /path/to/your/xero_script.py >> /var/log/xero_sync.log 2>&1
  

• Windows系统：添加任务计划程序，设置触发时间为每日夜间，执行动作选择你的Python解释器，参数填脚本路径即可。

注意事项 #

• 本地存储的token文件建议设置最小访问权限，避免泄露敏感数据
• 建议添加异常捕获和日志记录，方便排查定时任务执行失败的问题
• Xero接口有调用频率限制，拉取大量数据时建议添加适当的延迟，避免触发限流

内容的提问来源于stack exchange，提问作者Philippe Haumesser