CloudQuery Typeform 插件数据表解析：typeform_form_responses 表结构与增量同步原理

数据集成数据工程数据分析【免费下载链接】cloudqueryData pipelines for cloud config and security data. Build cloud asset inventory, CSPM, FinOps, and vulnerability management solutions. Extract from AWS, Azure, GCP, and 70 cloud and SaaS sources.项目地址https://gitcode.com/gh_mirrors/cl/cloudquery点击查看免费下载导读typeform_form_responses是 CloudQuery Typeform 源插件中用于采集 Typeform 表单提交响应数据的数据表它以typeform_forms为父表通过复合主键form_id, response_id唯一标识每条表单回复。本文将以该表的官方 schema 文档为骨架结合仓库中 表定义源码、父表定义、Typeform API 客户端 与 测试用例完整讲解表结构、父子关系、增量同步机制、底层 API 调用链与插件配置方式帮助你在实际同步任务中正确使用这张表。数据表概览该表的官方文档位于 typeform_form_responses.md核心定义如下表名Tabletypeform_form_responses中文语义TitleTypeform Form Responses即 Typeform 表单的提交响应复合主键Composite Primary Key由form_id与response_id两列共同组成保证某个表单 某条响应在目标数据库中全局唯一增量表Incremental从源码is_incrementalTrue可以看出该表支持增量同步只拉取新增或更新的响应避免全量重复拉取需要注意的是该表是一个从属表child table它依赖父表typeform_forms存在二者形成 1:N 的父子关系一个表单可以拥有多条响应记录。表结构与列说明完整列清单官方文档给出了该表的全部 12 个列及其数据类型这里逐列说明其业务含义列名类型SDK 表示含义form_id(PK)utf8所属表单的 ID取自已同步的父表记录的id字段response_id(PK)utf8Typeform 响应Response自身的 ID唯一标识一次提交landing_idutf8响应着陆页标识用于关联同一次会话的进入与提交landed_attimestamp[s]用户进入表单的时间戳秒级精度submitted_attimestamp[s]用户提交表单的时间戳秒级精度tokenutf8与该响应关联的 token 标识metadatajson响应元数据如浏览器、设备、来源等结构化信息answersjson用户在表单中的逐题答案集合hiddenjson表单中隐藏字段的取值calculatedjsonTypeform 计算字段calculated variables的结果variablesjson表单中定义的变量取值tagsjson附加在响应上的标签列表源码级列定义上述 schema 在源码 form_responses.py 中通过 CloudQuery Python SDK 的Column构造器定义与文档完全对应columns[ Column(form_id, pa.string(), primary_keyTrue), Column(response_id, pa.string(), primary_keyTrue), Column(landing_id, pa.string()), Column(landed_at, pa.timestamp(units)), Column(submitted_at, pa.timestamp(units)), Column(token, pa.string()), Column(metadata, JSONType()), Column(answers, JSONType()), Column(hidden, JSONType()), Column(calculated, JSONType()), Column(variables, JSONType()), Column(tags, JSONType()), ]这里的类型映射关系为文档中的utf8对应源码中的pa.string()Arrow 字符串类型文档中的timestamp[s]对应源码中的pa.timestamp(units)秒级时间戳文档中的json对应源码中的JSONType()SDK 提供的结构化 JSON 类型其中answers、metadata、hidden、calculated、variables、tags均采用 JSON 类型因为 Typeform API 返回的答案和元数据结构是动态的、可嵌套的JSON 类型可以在不改动表结构的前提下保留完整数据。父子表关系与数据模型从父表到子表的依赖链官方文档明确指出This table depends on typeform_forms.即typeform_form_responses依赖父表typeform_forms。反过来父表文档中也列出了依赖它的子表The following tables depend on typeform_forms:typeform_form_responses因此两张表的整体模型为typeform_forms1── 包含 ── typeform_form_responsesN父表typeform_forms的列包括id主键、created_at、last_updated_at、self、type、settings、theme、title、_links。源码中 forms.py 通过relations[FormResponses()]显式声明了这一父子关系class Forms(Table): def __init__(self) - None: super().__init__( nametypeform_forms, titleTypeform Forms, columns[...], relations[FormResponses()], )子表如何获得 form_id由于响应属于某个表单子表本身并不知道自己属于哪个表单。源码中的FormResponsesResolver.resolve()接收parent_resource父表记录在产出每条响应时把父表的id写入form_id字段form_response[form_id] parent_resource.item[id] yield form_response这也是form_id能作为复合主键组成部分的来源它由同步引擎在遍历父表时注入而非来自 Typeform API 的原始响应体。父表的FormsResolver通过child_resolvers属性挂载子表解析器确保同步时先枚举表单、再逐表单拉取其响应见 forms.py。增量同步原理基于 submitted_at 的游标typeform_form_responses是增量表is_incrementalTrue其增量逻辑完全体现在FormResponsesResolver.resolve()中form_responses.pydef resolve(self, client: Client, parent_resource: Resource): since self.state_client.get_key(typeform_form_responses_since) for form_response in client.client.list_form_responses( form_idparent_resource.item[id], sincesince, ): if not since or form_response[submitted_at] since: since form_response[submitted_at] self.state_client.set_key(typeform_form_responses_since, since) form_response[form_id] parent_resource.item[id] yield form_response其工作流程可以拆解为四步读取状态游标从状态客户端StateClient读取键typeform_form_responses_since首次同步时该值为空携带游标请求 API将since传给list_form_responses()让 Typeform API 只返回该时间点之后的响应推进游标逐条比较响应的submitted_at若大于等于当前游标则更新游标并通过state_client.set_key持久化注入父表 ID 并产出将父表单的id写入form_response[form_id]后交给同步调度器写入目标。值得注意的是游标键是全表共享的键名为typeform_form_responses_since而非按 form_id 分开存储同时游标推进逻辑位于循环内部、逐条比较保证在 API 返回结果乱序或存在边界情况时游标尽量单调递增。状态客户端由插件在sync()阶段构建并在同步结束后通过set_post_sync_hook(state_client.flush)统一落盘见 plugin.py因此增量进度可以跨多次同步持续生效。增量同步的适用前提增量能力依赖两个前提Typeform API 的/forms/{form_id}/responses端点支持since参数按响应提交时间过滤每条响应都携带submitted_at时间戳作为游标基准。若目标场景需要补拉历史数据或首次初始化直接以空since同步即可全量拉取后续同步会自动转为增量模式。底层 API 调用链与分页子表数据的真实来源是 Typeform API 的GET /forms/{form_id}/responses端点封装在 Typeform 客户端 中def list_form_responses(self, *, form_id, since, page1): params {page: page, page_size: 1000, since: since} resp self._get(f/forms/{form_id}/responses, paramsparams) ... resp resp.json() for form in resp[items]: yield form if resp[page_count] page: yield from self.list_form_responses( form_idform_id, sincesince, pagepage 1 )关键实现细节Bearer Token 认证所有请求通过Authorization: Bearer {access_token}请求头携带个人访问令牌client.py每页 1000 条page_size1000是响应列表的单页上限明显大于表单列表的page_size200以适配响应数据量大的场景递归分页根据响应体中的page_count判断是否还有下一页递归推进page直到拉完所有页错误处理非 200 状态码会抛出异常并携带 API 返回的原始错误文本便于排查鉴权或参数问题。调用链可以概括为FormResponsesResolver.resolve() └─ TypeformClient.list_form_responses(form_id, since) └─ GET https://api.typeform.com/forms/{form_id}/responses?page1page_size1000since...该行为在 test_client.py 中通过本地 Mock HTTP Server 得到验证test_list_forms_responses断言两次分页page_count2能分别取回两条响应记录且submitted_at值正确。而 test_forms.py 则通过 mock 客户端验证了同步typeform_forms会连带迁移并写入typeform_form_responses两张表的完整链路。同步该表的插件配置要真正同步typeform_form_responses需要在 CloudQuery 配置文件中以 source 插件方式声明 Typeform 源。参考 testdata/config.yml 与 官方配置文档一个最小可用配置如下kind: source spec: name: typeform registry: docker path: docker.cloudquery.io/cloudquery/source-typeform:VERSION_SOURCE_TYPEFORM tables: [typeform_forms] # 选择父表即自动包含子表响应 destinations: [DESTINATION_NAME] spec: access_token: ${TYPEFORM_ACCESS_TOKEN} # base_url: https://api.typeform.com # 可选EU 账号用 https://api.eu.typeform.com # concurrency: 100 # 可选默认 100 # queue_size: 10000 # 可选默认 10000要点说明表选择官方配置示例以tables: [typeform_forms]作为入口。由于父子关系选择父表会连带同步子表typeform_form_responses也可显式列出[typeform_forms, typeform_form_responses]或用[*]同步全部表认证access_token为必填项来自 Typeform Dashboard 的个人访问令牌建议创建只读权限令牌见 认证文档。未提供时插件会直接抛出access_token must be provided异常client.pybase_url默认https://api.typeform.com若账号数据存储于欧盟应改为https://api.eu.typeform.comconcurrency / queue_size控制同步调度器的并发请求数与队列容量默认分别为 100 与 10000二者会注入 CloudQuery Python SDK 的Scheduler见 plugin.py运行环境Typeform 插件以 Docker 镜像形式分发需要本地安装 Docker 运行时且 CloudQuery CLI 需为支持dockerregistry 类型的版本v3.12.0 及以上。目标端只需是任意 CloudQuery 支持的 destination如 PostgreSQL、BigQuery、Snowflake 等。form_id与response_id构成的复合主键会在目标库中落为联合主键约束重复同步时通过主键去重更新。常见使用场景与注意事项全量初始化后自动增量首次同步不携带since拉取全部历史响应并记录最新submitted_at游标之后的同步只拉取新增响应节省 API 配额与目标端写入压力。复合主键的运维意义因为同一表单的response_id全局唯一复合主键主要保证表单 响应联合维度上的幂等写入跨表单时response_id允许重复勿将其误当作全局主键使用。JSON 列的查询方式answers、metadata等列以 JSON 类型存储在支持 JSON 函数的目标库如 PostgreSQL、ClickHouse中可直接用 JSON 路径表达式展开分析答案明细在普通 SQL 数据库中它们会被当作结构化字符串存储。游标的粒度为全表当前实现用单一键typeform_form_responses_since记录全表进度form_responses.py若多个表单的响应提交时间跨度很大增量进度以最近一次拉取到的响应时间为准。表单列表与响应列表分页上限不同表单接口每页 200 条响应接口每页 1000 条client.py对超大量表单或响应均有递归分页兜底不会漏数据。延伸阅读父表文档typeform_forms.md表格索引tables/README.md插件总览与配置参考overview.md、_configuration.md表定义源码form_responses.py、forms.pyAPI 客户端与同步调度plugin/typeform/client.py、plugin/plugin.py测试验证tests/tables/test_forms.py、tests/typeform/test_client.py赞分享数据集成数据工程数据分析【免费下载链接】cloudqueryData pipelines for cloud config and security data. Build cloud asset inventory, CSPM, FinOps, and vulnerability management solutions. Extract from AWS, Azure, GCP, and 70 cloud and SaaS sources.项目地址https://gitcode.com/gh_mirrors/cl/cloudquery点击查看免费下载相关推荐CloudQuery Square 源插件数据表解析square_payouts 表结构与同步实战CloudQuery Square 源插件数据表解析square_payouts 表结构与同步实战 square_payouts 是 CloudQuery S数据集成数据工程数据分析CloudQuery Square 源插件表结构解析square_invoices 发票数据模型与同步原理CloudQuery Square 源插件表结构解析square_invoices 发票数据模型与同步原理 square_invoices 是 CloudQu数据集成数据工程数据分析CloudQuery Hacker News 源插件增量表详解hackernews_items 的表结构、游标同步原理与配置实战CloudQuery Hacker News 源插件增量表详解hackernews_items 的表结构、游标同步原理与配置实战 导读 hackernews_数据集成数据工程数据分析上一篇在 VS Code 与 cn CLI 上运行 Continue.dev 循环工程primitive 映射、无头调度与 maker/checker 实践下一篇Dopamine 中的 AtariPreprocessingAtari 2600 图像预处理类源码级解析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考