Coinbase Python API 上的分页

Posted 2023-02-18

【中文标题】Coinbase Python API 上的分页

【英文标题】：Pagination on Coinbase Python API

【发布时间】：2017-11-05 04:04:41

【问题描述】：

我正在尝试获取 Coinbase 帐户上的所有交易，这需要分页。 documentation 很少介绍如何在 Python 中执行此操作，但我已经设法使其工作：

client = Client(keys['apiKey'], keys['apiSecret'])
accounts = client.get_accounts()

for account in accounts.data:
    txns = client.get_transactions(account.id, limit=25)
    while True: 
        for tx in txns.data:
            print(tx.id)

        if txns.pagination.next_uri != None:
            starting_after_guid = re.search('starting_after=([0-9a-f]8-([0-9a-f]4-)3[0-9a-f]12)', txns.pagination.next_uri, re.I)[1]
            txns = client.get_transactions(account.id, limit=25, starting_after=starting_after_guid)
        else:
            break

分页对象仅包含 next_uri 其他所有内容均为 null/None——它应该包含一个包含 starting_after 以及其他有用数据的字典。正则表达式搜索看起来很傻，但确实有效。

有没有更好的方法？

【参考方案1】：

这就是它可以在没有正则表达式的情况下工作的方式：

_cb = Client(keys['apiKey'], keys['apiSecret'])
accounts = client.get_accounts()

for account in accounts.data:
    """ Gets the transactions history from coinbase """
    all_txns = []
    starting_after = None
    while True:
        txns = _cb.get_transactions(account['id'], limit=100, starting_after=starting_after)
        if txns.pagination.next_starting_after is not None:
            starting_after = txns.pagination.next_starting_after
            for tx in txns.data:
                all_txns.append(tx)
            time.sleep(1)  # Let's not hit the rate limiting
        else:
            for tx in txns.data:
                all_txns.append(tx)
            break

请注意，next_starting_after 取自事务并用作下一个查询的 starting_after 参数。

【参考方案2】：

这只是让其他答案更进一步，并创建了一个通用函数，因此您可以将它用于 api 中的大多数分页方法。

from coinbase.wallet.client import Client


client = Client(API_KEY, SECRET)

def _get_paginated_items(api_method, limit=100):
    """Generic getter for paginated items"""
    all_items = []
    starting_after = None
    while True:
        items = api_method(limit=limit, starting_after=starting_after)
        if items.pagination.next_starting_after is not None:
            starting_after = items.pagination.next_starting_after
            all_items += items.data
        else:
            all_items += items.data
            break
    return all_items


def get_accounts(client, limit=100):
    return _get_paginated_items(client.get_accounts, limit)


def get_transactions(account, limit=100):
    return _get_paginated_items(account.get_transactions, limit)

# Use them.
accounts = get_accounts(client)
for account in accounts:
    transactions = get_transactions(account)
    # etc...

【参考方案3】：

sn-p 上面遇到错误。这行得通

client = Client(keys['apiKey'], keys['apiSecret'])
accounts = client.get_accounts()

for account in accounts.data:
    txns = client.get_transactions(account.id, limit=25)
    while True:
        for tx in txns.data:
            print(tx.id)

        if txns.pagination.next_uri != None:
            starting_after_guid = re.search('starting_after=(.*)', str(txns.pagination.next_uri), re.I).group(1)
            txns = client.get_transactions(account.id, limit=25, starting_after=starting_after_guid)
        else:
            break

【参考方案4】：

这是我的迭代器代码：

from coinbase.wallet.client import Client


API_KEY = '***************'
API_SECRET = '***************************'


def get_accounts_iterator(client, limit=25):
    starting_after = None
    
    while True:
        accounts = client.get_accounts(limit=limit, starting_after=starting_after)
        for account in accounts.data:
            yield account
        
        if accounts.pagination.next_starting_after is None:
            break
        
        starting_after = accounts.pagination.next_starting_after


def get_transactions_iterator(client, account_id, limit=25):
    starting_after = None
    
    while True:
        txs = client.get_transactions(account_id, limit=limit, starting_after=starting_after)
        for tx in txs.data:
            yield tx
        
        if txs.pagination.next_starting_after is None:
            break
        
        starting_after = txs.pagination.next_starting_after


def main():
    client = Client(API_KEY, API_SECRET)
    
    for account in get_accounts_iterator(client):
        if float(account.get('balance',).get('amount',0))==0:
            continue
        
        print(account.balance.currency, account.balance.amount)
        
        for tx in get_transactions_iterator(client, account.id):
            print(tx.amount.amount)


if __name__ == '__main__':
    main()