飞书API(3):Python 自动读取多维表所有分页数据的三种方法

上一小节介绍了怎么使用 Python 读取多维表的数据,看似可以成功获取到了所有的数据,但是在实际生产使用过程中,我们会发现,上一小节的代码并不能获取到所有的多维表数据,它只能获取一页,默认是第一页。因为我使用的测试表数据量小,第一页便可读取完,所以有一种获取到所有数据的错觉。
本文介绍如何进行分页读取飞书多维表的所有数据,主要内容介绍如下:

  • 在线测试:使用飞书 API 调试台测试
  • 本地测试:使用 Python 代码测试,介绍通过三种不同的方法自动读取所有分页数据:通过 while 循环读取、通过 for 循环读取和通过内函数递归读取。

1、如何读取所有分页数据

那么怎么获取到所有的数据呢?
查看官方文档-查询记录,在“查询参数”中(如下图),可以看到有2个和翻页相关的参数:“page_token”和“page_size”。

image.png

先说下“page_size”,和分页记录数相关,默认值为 20,最大值上图没介绍,不过在 API 开头有说明(如下图),单次最大查询 500 行记录。

image.png

而“page_token”是一个翻页参数,首页没有该参数,从第 2 页开始就必须要传递相关的参数,而相关的“page_token”参数的值需要通过上一页返回的数据(即响应体)进行获取,也就是说从第 2 页开始都依赖上一页的返回。
分页标记,第一次请求不填,表示从头开始遍历;分页查询结果还有更多项时会同时返回新的 page_token,下次遍历可采用该 page_token 获取查询结果。

查看响应体说明,可以看到有两个参数和翻页相关:“has_more”和“page_token”。

  • “has_more”用于判断有没有更多的页,有则返回 True。
  • “page_token”用于传递给查询数据接口,请求下一页的数据。

image.png

1.1 调试:读取第二页数据

了解了取数逻辑之后,可以先调试一下。调试逻辑:先获取第 1 页,然后获取到第 2页的“page_token”再发起请求。

1.1.1 读取第 1 页

打开 API 调试台,准备多维表,多维表的参数说明等,参考 第 2 小节:《API(2):通过 Python 读取多维表数据节》的【1.1 获取 demo】。

  • 请求头 和 路径参数参考第 2 小节填写即可
  • 查询参数:第 2 小节只填写了必填参数,所以查询查数没有传递值(即使用默认参数值),由于我测试的数据仅有 10 条记录,所以“page_size”改为 5,以便观察翻页的参数值。

image.png
改完直接点击【开始调试】即可,结果如下,可以看到“has_more”的值为 true,“page_token”返回一个字符串。
image.png
注:数据在响应体的 data>items 中,可以展开和多维表的记录比对查看下。

1.1.2 读取第 2 页

将首次请求返回的“page_token”的值填写到路径参数中“page_token”的输入框里,再次点击【开始调试】发起请求。
image.png

结果如下,可以看到“has_more”为 false,没有“page_token”参数。10 条记录分两次取完了,所以这个结果符合预期。
image.png

1.2 调试:本地测试读取第二页数据

续上,第二次请求结果,点击示例代码,点击“Python-requests”,可以看到 API 连接多了 3 个参数,复制代码到本地测试运行。
image.png

测试结果如下,正常获取到第二页的数据。

image.png

2、使用 Python 自动读取所有分页数据

上面通过手动输入“page_token”的方式读取到了测试文档的所有数据,在实际生产过程中,不可能手动一页页处理,所以需要通过代码自动将第一次获取的“page_token”传递给第二次请求。

根据飞书提供的信息,至少我们可以有三种方法来穷尽读取任意多维表的所有数据。

  • 方法一:通过 while 循环读取
  • 方法二:通过 for 循环读取
  • 方法二:通过内函数递归读取

上一小节最后有一个获取tenant_access_token并调接口读取数据的合并代码(代码如下)。本次测试基于该代码进行迭代。

import requests
import json

def get_tenant_access_token(app_id, app_secret):
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = json.dumps({
        "app_id": app_id,
        "app_secret": app_secret
    })
    headers = {'Content-Type': 'application/json'}
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()['tenant_access_token']

def get_bitable_datas(tenant_access_token, app_token, table_id):
    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search"
    payload = json.dumps({})
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {tenant_access_token}'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    print(response.text)
    
app_id = 'your_app_id'
app_secret = 'your_app_secret'
tenant_access_token = get_tenant_access_token(app_id, app_secret)
app_token = 'your_app_token'
table_id = 'your_table_id'
get_bitable_datas(tenant_access_token, app_token, table_id)

2.1 while 循环读取分页数据

思路:主代码使用while循环调用get_bitable_datas()函数,并处理提取page_token,在下一次循环传递给get_bitable_datas()函数。
get_bitable_datas()函数需要进行修改,以便适配任意页的读取,并返回调用接口的数据。

  • url新增三个参数:page_sizepage_tokenuser_id_type;第一页page_token=''可以正常请求数据;
  • 返回调用接口的数据:加上return response.json()

修改完参考如下:

def get_bitable_datas(tenant_access_token, app_token, table_id, page_token='', page_size=20):

    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search?page_size={page_size}&page_token={page_token}&user_id_type=user_id"
    payload = json.dumps({})
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {tenant_access_token}'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()

主代码加上while循环,把调用get_bitable_datas()函数放到循环体中,并提取返回数据中的page_tokenhas_more,打印记录。参考如下:

app_id = 'your_app_id'
app_secret = 'your_app_secret'
tenant_access_token = get_tenant_access_token(app_id, app_secret)
app_token = 'your_app_token'
table_id = 'your_table_id'

page_token = ''
page_size = 5
has_more = True
while has_more:
    response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size)
    if response['code'] == 0:
        page_token = response['data'].get('page_token')
        has_more = response['data'].get('has_more')
        print(response['data'].get('items'))

运行代码,结果如下:成功读取并打印分页的数据。

在这里插入图片描述

将主代码封装到main()函数中,并将所有返回的结果整到一个列表中,最终参考代码如下:

import requests
import json

def get_tenant_access_token(app_id, app_secret):
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = json.dumps({
        "app_id": app_id,
        "app_secret": app_secret
    })
    headers = {'Content-Type': 'application/json'}
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()['tenant_access_token']

def get_bitable_datas(tenant_access_token, app_token, table_id, page_token='', page_size=20):

    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search?page_size={page_size}&page_token={page_token}&user_id_type=user_id"
    payload = json.dumps({})
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {tenant_access_token}'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()

def main():
    app_id = 'your_app_id'
    app_secret = 'your_app_secret'
    tenant_access_token = get_tenant_access_token(app_id, app_secret)
    app_token = 'your_app_token'
    table_id = 'your_table_id'

    page_token = ''
    page_size = 5
    has_more = True
    feishu_datas = []
    while has_more:
        response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size)
        if response['code'] == 0:
            page_token = response['data'].get('page_token')
            has_more = response['data'].get('has_more')
            # print(response['data'].get('items'))
            # print('\n--------------------------------------------------------------------\n')
            feishu_datas.extend(response['data'].get('items'))
        else:
            raise Exception(response['msg'])
        
    return feishu_datas

if __name__ == '__main__':
    feishu_datas = main()
    print(feishu_datas)

执行结果如下:

在这里插入图片描述

2.2 for 循环读取分页数据

注:在 2.1 基础上修改。
使用 for 循环,不如 while 循环便捷。因为 for 循环有一个难点,就是需要先确认循环的次数。而确认次数,需要知道总数据量。而确认总数据量,需要通过调接口获取,每次查询请求都会返回一个记录数据量的参数:total (如下图)。
image.png

读取该值后结合每页的记录数page_size,使用公式total/size-1,然后向上取整来确定循环的次数。
关联核心代码如下:

  • 先调一次接口获取到 total 的参数值,然后根据公式计算需要循环多少次(需要使用math.ceil()方法向上取整);顺带将首次请求的page_tokenfeishu_datas处理
  • 然后根据上面得到的次数进行遍历,读取后面页数的数据,并拼接到feishu_datas中。
    # 首次请求单独处理,获取 total、page_token、has_more、items
    response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size)
    if response['code'] == 0:
        total = math.ceil(response['data'].get('total')/page_size - 1)
        page_token = response['data'].get('page_token')
        feishu_datas = response['data'].get('items')
        # 非首次请求,统一处理
        for i in range(total):
            response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size)
            if response['code'] == 0:
                page_token = response['data'].get('page_token')
                feishu_datas.extend(response['data'].get('items'))

最终的完整参考代码如下:

  • 首次请求和非首次请求的数据分开处理;
  • 使用 total 计算循环次数,不需要 has_more 参数。
import requests
import json, math

def get_tenant_access_token(app_id, app_secret):
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = json.dumps({
        "app_id": app_id,
        "app_secret": app_secret
    })
    headers = {'Content-Type': 'application/json'}
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()['tenant_access_token']

def get_bitable_datas(tenant_access_token, app_token, table_id, page_token='', page_size=20):

    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search?page_size={page_size}&page_token={page_token}&user_id_type=user_id"
    payload = json.dumps({})
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {tenant_access_token}'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()

def main():
    app_id = 'your_app_id'
    app_secret = 'your_app_secret'
    tenant_access_token = get_tenant_access_token(app_id, app_secret)
    app_token = 'your_app_token'
    table_id = 'your_table_id'

    page_token = ''
    page_size = 5
    # 首次请求单独处理,获取 total、page_token、has_more、items
    response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size)
    if response['code'] == 0:
        total = math.ceil(response['data'].get('total')/page_size - 1)
        page_token = response['data'].get('page_token')
        feishu_datas = response['data'].get('items')
        # 非首次请求,统一处理
        for i in range(total):
            response = get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size)
            if response['code'] == 0:
                page_token = response['data'].get('page_token')
                feishu_datas.extend(response['data'].get('items'))
            else:
                raise Exception(response['msg'])
    else:
        raise Exception(response['msg'])
        
    return feishu_datas

if __name__ == '__main__':
    feishu_datas = main()
    print(feishu_datas)

执行结果如下:
在这里插入图片描述

2.3 内函数递归读取分页数据

注:在 2.1 基础上修改。
本小节介绍使用内函数进行递归处理,相对会比较复杂。
目标是传入相关的参数,便可获取到多维表的所有数据。
实现该目标主要是在get_bitable_datas()函数中进行迭代:

  • 递归逻辑:加上内函数inner(),在内函数中调用外函数get_bitable_datas()
  • 什么时候调用内函数?has_moreTrue的时候。

数据的存储问题:将存储数据的列表feishu_datas放到函数外作为全局变量,函数get_bitable_datas()的功能是给列表feishu_datas添加元素。

直接看看最终完整的参考代码:

  • feishu_datas变量放到函数之前,使得全局可调用
  • get_bitable_datas()中新增内函数inner(),该函数功能只有一个就是调用外函数get_bitable_datas()。当has_more值为True的时候,便不断调用内函数inner(),也不断调用外函数get_bitable_datas(),直到最后一页has_more值为False时,一层层返回,最终得到记录所有数据的列表feishu_datas
  • main()函数中直接调用get_bitable_datas()函数便可得到记录所有数据的列表feishu_datas
import requests
import json

feishu_datas = []

def get_tenant_access_token(app_id, app_secret):
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = json.dumps({
        "app_id": app_id,
        "app_secret": app_secret
    })
    headers = {'Content-Type': 'application/json'}
    response = requests.request("POST", url, headers=headers, data=payload)
    # print(response.text)
    return response.json()['tenant_access_token']

def get_bitable_datas(tenant_access_token, app_token, table_id, page_token='', page_size=20):

    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search?page_size={page_size}&page_token={page_token}&user_id_type=user_id"
    payload = json.dumps({})
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {tenant_access_token}'
    }
    response = requests.request("POST", url, headers=headers, data=payload).json()

    def inner(tenant_access_token, app_token, table_id, page_token):
        # 调用外函数,递归调用
        get_bitable_datas(tenant_access_token, app_token, table_id, page_token)

    if response['code'] == 0:
        page_token = response['data'].get('page_token')
        has_more = response['data'].get('has_more')
        feishu_datas.extend(response['data'].get('items'))
        if has_more:
            # 调用内函数
            inner(tenant_access_token, app_token, table_id, page_token)
    else:
        raise Exception(response['msg'])

def main():
    app_id = 'your_app_id'
    app_secret = 'your_app_secret'
    tenant_access_token = get_tenant_access_token(app_id, app_secret)
    app_token = 'your_app_token'
    table_id = 'your_table_id'

    page_token = ''
    page_size = 5
    get_bitable_datas(tenant_access_token, app_token, table_id, page_token, page_size)

if __name__ == '__main__':
    main()
    print(feishu_datas)

执行结果如下:

在这里插入图片描述

3、小结

使用 Python 读取多维表分页,需要传递“page_token”参数。
第一页“page_token”参数为空字符串,第二页开始从上一页的响应体提取“page_token”的值。
使用 Python 实现,可以通过三种方式读取所有分页的数据:while 循环读取、for 循环读取和内函数递归读取。

  • while 循环:使用has_more参数,直到has_more=False跳出循环;
  • for 循环:使用total参数,结合page_size计算循环次数;
  • 内函数递归:使用has_more参数递归调用内函数,直到has_more=False时逐层返回结果。

while 循环 相对比较好理解,而且很简洁。for 循环通过数量进行计算,或有一定风险,就是在读取的过程中,如果新增数据,可能会出现 bug,如果新增数据涉及新的页,将读取不到新增的数据。内函数递归方法,理解难度相对较高,代码相对简洁。
个人推荐顺序:while 循环>内函数递归>for 循环。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/525371.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

MySQL的存储引擎、索引与事务

常见的端口号 MySQL–3306http–80https–443tcp–23fcp–21tomcat–8080ssh–22oracle–1521rockermq–9876 存储引擎 使用指令查看所有引擎: show engines;从图中可以看出MySQL默认的存储引擎是InnoDB;并且在5.7版本所有的存储引擎中只有 InnoDB 是…

亮数据----教你轻松获取数据

文章目录 1. 数据采集遇到的瓶颈1.1 不会造数据?1.2 不会写爬虫代码? 2.IP 代理基础知识2.1 基本概念2.2 作用和分类2.3 IP 代理的配置和使用2.4 安全和合规 3. 为何使用亮数据 IP 代理3.1 拥有丰富的代理网络服务3.2 简单易操作的采集工具3.3 拥有各平台…

路由器对数据包的处理过程分析笔记

虽然TCP-IP协议中传输数据会在各个路由器再次经过物理层、链路层、网络层的解封装、加工、封装、转发,但是对于两个主机间的运输层,在逻辑上,应用进程是直接通信的。 路由器主要工作在网络层,但它也涉及到物理层和链路层的一些功能…

PWM 脉宽跟随方案介绍

1. 前言 数字电源产品在使用桥式电路拓扑或是多路交错控制中,有时会需要滞后臂的 PWM 脉宽严格跟随超前臂的 PWM 脉宽,或从路的 PWM 脉宽严格跟随主路的 PWM 脉宽,本文将介绍如何利用高精度定时器实现 PWM 输出脉宽跟随,一种使用…

ai智能电销机器人的核心技术,工作原理和作用

科技快速发展的同时,带来了人工智能产品的普及。而ai智能电销机器人则成为推进电销行业的产物,那么ai智能电销机器人是如何帮助企业高效触客,有效地工作,效果又如何呢?我们一起来看看吧! 一、ai智能电销机器…

软件的生命周期_瀑布模型

瀑布模型 描述软件生成到消亡的过程模型图 该模型目前实际工作中已不常用,但是该模型是其他新型模型的“鼻祖 瀑布模型的优点 每个阶段比较清楚,并且有对应的文档产生 当前一个阶段完成后,才开始后面的阶段(一次性的&#xff09…

「媒体邀约」天津媒体邀约资源有哪些?媒体宣传现场报道

传媒如春雨,润物细无声,大家好,我是51媒体网胡老师。 天津作为中国北方的重要城市,拥有丰富的媒体资源,可以为各类活动提供全面的媒体宣传和现场报道。以下是天津地区的媒体邀约资源: 1. 报纸媒体 - 《天…

如何搭建APP分发平台分发平台搭建教程

搭建一个APP分发平台可以帮助开发者更好地分发和管理他们的应用程序。下面是一个简要的教程,介绍如何搭建一个APP分发平台。 1.确定需求和功能:首先,确定你的APP分发平台的需求和功能。考虑以下几个方面: 用户注册和登录&#xff…

Redis从入门到精通(九)Redis实战(六)基于Redis队列实现异步秒杀下单

文章目录 前言4.5 分布式锁-Redisson4.5.4 Redission锁重试4.5.5 WatchDog机制4.5.5 MutiLock原理 4.6 秒杀优化4.6.1 优化方案4.6.2 完成秒杀优化 4.7 Redis消息队列4.7.1 基于List实现消息队列4.7.2 基于PubSub的消息队列4.7.3 基于Stream的消息队列4.7.4 基于Stream的消息队…

DIY可视化UniApp表格组件

表格组件在移动端的用处非常广泛,特别是在那些需要展示结构化数据、进行比较分析或提供详细信息的场景中。数据展示与整理:表格是展示结构化数据的理想方式,特别是在需要展示多列和多行数据时。通过表格,用户可以轻松浏览和理解数…

Spring Boot-02-依赖管理和自动配置

二、Spring Boot的两大重要机制 1. 依赖管理机制 开发什么场景,导入什么场景启动器,场景启动器自动把这个场景的所有核心依赖全部导入进来。maven依赖传递原则:A依赖B,B依赖C,则A就拥有B和C。每个boot项目都有一个父…

防火墙配置IPSec VPN【IPSecVPN概念及详细讲解】

防火墙配置IPSecVPN-点到点 配置目标 总公司内网与分公司内网互通 拓扑 配置ISP路由器 <Huawei>u t m <Huawei>sys [Huawei]sys ISP [ISP]interface g0/0/0 [ISP-GigabitEthernet0/0/0]ip address 100.1.1.102 24 [ISP-GigabitEthernet0/0/0]q [ISP]interface g…

【cocos creator】【编辑器插件】cocos creator文件复制时,解决cocos creator uuid冲突

&#xff01;&#xff01;&#xff01;修改前先备份 1、将文件夹放在packages文件夹下 2、打开项目&#xff0c;选择要刷新uuid的文件夹 3、菜单栏点击 扩展->refresh-uuid 4、等控制台提示&#xff1a;资源uuid刷新完成&#xff0c;重启项目&#xff08;&#xff01;&#…

VSCODE自动更新无法连接远程服务器报错“waiting for server log...“的解决方法

问题描述 一觉醒来打开vscode发现连接远程服务器显示无法连接&#xff0c;终端一直报错“waiting for server log…"&#xff0c;经查是因为vscode自动更新到了1.86&#xff0c;对于远程服务器的linux版本要求较高。这里记录下解决方法。 解决方法 1. 下载vscode便携版…

网络安全之命令注入

漏洞原理&#xff1a; 应用系统设计需要给用户提供指定的远程命令操作的接口&#xff0c;比如&#xff1a;路由器&#xff0c;防火墙&#xff0c;入侵检测等设备的web管理界面。一般会给用户提供一个ping操作的web界面 用户从web界面输入目标IP&#xff0c;提交后台会对改IP地…

Web 前端性能优化之五:构建优化

4、构建优化 资源的合并与压缩所涉及的优化点包括两方面&#xff1a;一方面是减少HTTP的请求数量&#xff0c;另一方面是减少HTTP请求资源的大小。 1、HTML 压缩 1、什么是 HTML 压缩 百度首页部分 HTML 源代码 谷歌首页部分 HTML 源代码 虽然这些格式化的字符能带来很好的代…

LiDAR点云转3D Tiles

到 2026年&#xff0c;一个国家项目旨在绘制整个法国领土的三维地图。 美国国家地理和林业信息研究所 (IGN) 是这个名为“Lidar HD”的“项目”的幕后黑手&#xff0c;其目标是获得该地区非常精确的 3D 描述。 NSDT工具推荐&#xff1a; Three.js AI纹理开发包 - YOLO合成数据生…

内外网数据交换发展进程:安全与便捷并行

随着信息化的不断推进&#xff0c;医院、党政以及企业的内外网数据交换正成为日益关注的焦点。在保障数据安全的前提下&#xff0c;需要寻求一种既安全可靠又操作便捷的数据传输方式。本文将探讨内外网数据交换发展进程&#xff0c;分析各种传输方式的优缺点&#xff0c;以及它…

爬虫入狱笔记——xx政府网站公开政策数据

最近在学习爬虫&#xff0c;做个笔记吧 今天爬xx政府网站-政策法规栏目的数据 咱们首先需要找到数据从哪里来&#xff0c;鼠标右键->检查&#xff08;或者快捷键一般为F12&#xff09;检查元素&#xff0c;搜索关键词 eg.【违法案例】 回车&#xff0c; 如果没有的话&am…

【随笔】Git 高级篇 -- 整理提交记录(下)rebase(十六)

&#x1f48c; 所属专栏&#xff1a;【Git】 &#x1f600; 作  者&#xff1a;我是夜阑的狗&#x1f436; &#x1f680; 个人简介&#xff1a;一个正在努力学技术的CV工程师&#xff0c;专注基础和实战分享 &#xff0c;欢迎咨询&#xff01; &#x1f496; 欢迎大…