在 Mage 中配置 Knowi 数据源：连接认证、配置参数与数据抽取机制详解-平芜编程栈

数据工程
数据编排
ETL
任务调度
批处理
流处理
数据集成
后端

【免费下载链接】mage-ai

🧙 Build, run, and manage data pipelines for integrating and transforming data.

项目地址：https://gitcode.com/gh_mirrors/ma/mage-ai

Knowi 是一款 BI 与数据可视化平台，其 Management API 提供了对 Dashboard（仪表盘）和 Widget（组件）等元数据的访问能力。Mage 通过内置的 Knowi 数据源（Source），可以基于 Bearer Token 认证将 Knowi 中的仪表盘与组件元数据抽取进数据管道。本指南以开源仓库中 Knowi 数据源文档为骨架，结合源码实现与 Stream 定义，完整讲解配置参数、access_token获取方式、可用数据流以及底层抽取与重试机制，帮助你在 Mage 中快速落地 Knowi 数据集成。

一、Knowi 数据源概览

Knowi 数据源是 Mage 内置的众多数据集成 Source 之一，在mage_ai/data_integrations/sources/constants.py中与 Airtable、Salesforce、Stripe 等一同登记为官方支持的接入源。整个 Source 的实现位于：

入口与抽取逻辑：mage_integrations/mage_integrations/sources/knowi/init.py
API 客户端：mage_integrations/mage_integrations/sources/knowi/client.py
数据流（Stream）定义：mage_integrations/mage_integrations/sources/knowi/streams.py
Schema 加载：mage_integrations/mage_integrations/sources/knowi/schema.py
字段级 Schema：schemas/dashboards.json 与 schemas/widgets.json

从源码结构看，该 Source 遵循 Singer 协议的 tap 风格：通过load_data方法按 Stream 逐个拉取记录，每个 Stream 决定自身的复制方法（Replication Method）、主键（Key Properties）与合法复制键（Valid Replication Keys）。

二、配置参数详解

配置 Knowi 数据源时需要提供以下凭证与参数。下表完整继承自 README 配置表，并与 templates/config.json 中声明的字段一一对应：

Key	描述	示例值	必填
`access_token`	用于认证 Knowi 账号的访问令牌（Bearer Token）	`abcdefg123456`	是（REQUIRED）
`request_timeout`	请求超时时间（秒），默认 300 秒	`300`	否
`user_agent`	随请求头发送的用户代理（User Agent）字符串	`my-app-v1.0`	否
`start_date`	过滤结果：使用增量同步（Incremental Sync）时，仅拉取在`start_date`之后更新的记录	`2023-01-01`	否

在 Mage 的数据集成配置界面中，这些参数会被写入连接配置，最终作为config字典传给 Source 实例。templates/config.json提供了标准的配置模板骨架：

{ "access_token": "", "start_date": null, "request_timeout": null, "user_agent": null }

1.`access_token`（必填）：Knowi API 的认证凭据

这是唯一必填参数。Knowi 采用 Bearer Token 认证，客户端在每次请求的 HTTP 头中携带Authorization: Bearer <access_token>。获取方式请遵循 Knowi 官方文档中Management API 的 API Authentication一节给出的指引：在 Knowi 账号的 API 管理页面中创建并获取访问令牌，然后在 Mage 的 Knowi Source 配置中将该令牌填入access_token字段。

从源码看，KnowiClient 构造函数会接收access_token并保存；如果该值为None，check_access_token 会直接抛出Error: Missing access_token.异常。同时，当 Knowi 返回 HTTP 401 且错误码中包含access_token时，客户端会明确提示当前令牌已过期或无效，要求重新认证以生成新的令牌后继续抽取（见 raise_for_error）。

安全建议：将access_token视为敏感凭据，在 Mage 中可通过密钥管理（Secrets）机制引用环境变量，避免明文写入代码仓库。

2.`request_timeout`（可选）：请求超时控制

指定单个 HTTP 请求的超时秒数，默认值为 300 秒。在 KnowiClient 初始化逻辑中，超时值按如下规则解析：

若config_request_timeout存在且float(config_request_timeout)为真（即非 0、非空字符串、非"0"），则采用该值；
否则回退到模块级常量REQUEST_TIMEOUT = 300。

也就是说，传入0、"0"或空值都会被当作"未配置"，最终使用 300 秒默认值。该超时值随后在check_access_token与request方法中以timeout=self.__request_timeout传入底层requests.Session调用。

3.`user_agent`（可选）：请求标识

自定义的 User Agent 字符串，会写入每次请求的User-Agent请求头，便于 Knowi 侧识别调用方应用，例如my-app-v1.0。源码中，只要user_agent非空，就会在 check_access_token 与 request 的请求头中注入。

4.`start_date`（可选）：增量同步的起始时间点

当 Stream 采用增量复制（INCREMENTAL）时，start_date用于过滤记录，仅同步该时间点之后更新的数据。在 load_data 中可以看到具体逻辑：从 bookmark 中优先读取updated_at作为断点；若 bookmark 不存在，则回退到配置中的start_date。bookmark 为整数时按 Unix 时间戳解析，否则用singer.utils.strptime_to_utc解析为 UTC 时间。格式建议为YYYY-MM-DD（如2023-01-01）。

三、如何获取`access_token`

根据 README 与官方文档站对应页面（docs/data-integrations/sources/knowi.mdx）的说明：

登录 Knowi 账号；
进入Management API相关的 API 认证（API Authentication）页面；
按 Knowi 官方指引创建 API Token；
将生成的令牌复制到 Mage 中 Knowi Source 的access_token配置项。

需要注意的是，Knowi 的访问令牌存在有效期/失效机制。源码在捕获 401 响应时会记录如下日志，提醒你重新认证并刷新令牌后继续同步：

Your API access_token is expired/invalid as per Knowi's security policy. Please re-authenticate your connection to generate a new access_token and resume extraction.

四、可用的数据流（Streams）与复制方式

Knowi 数据源共暴露 4 个 Stream，全部定义在 streams.py 的STREAMS字典中：

Stream ID	复制方式	主键	说明
`dashboard_list`	FULL_TABLE	`id`	仪表盘列表，仅作为父 Stream 供子 Stream 消费，不参与同步
`dashboards`	FULL_TABLE	`id`	仪表盘详情，逐条拉取每个仪表盘对象
`widget_list`	FULL_TABLE	`id`	组件列表，仅作为父 Stream 供子 Stream 消费，不参与同步
`widgets`	FULL_TABLE	`id`	组件详情，逐条拉取每个组件对象

值得注意的实现细节：

父子 Stream 设计：DashboardList与WidgetList的to_replicate = False，它们不会作为独立表被同步，而是通过 get_parent_data 向子 Stream 提供父级 ID 列表。例如 Dashboards.get_records 先拿到所有dashboard_id，再逐个请求dashboards/{id}组装详情。
复制方式：当前 4 个 Stream 均为FULL_TABLE（全量复制）。这与start_date的语义对应——start_date仅在 Stream 的复制方式为 INCREMENTAL 时生效；全量同步下bookmark_datetime会被置为None（见 load_data）。
响应数据键：Knowi API 列表类响应统一放在list键下（default_data_key = "list"），详情类数据则使用各自的data_key（如dashboards、widgets）。

每个 Stream 的主键、复制键与复制方式会通过 get_forced_replication_method、get_table_key_properties、get_valid_replication_keys暴露给 Mage 编排层，用于决定目标表的 Schema、主键与增量策略。字段级 Schema 则由 schema.py 在 Discovery 阶段从schemas/*.json加载，并通过 Singer metadata 标注key_properties、valid_replication_keys与replication_method。

以dashboards为例，schemas/dashboards.json 声明的字段包括：id（整数，主键）、name、url、createdDate、lastModDt、lastAccessDt、userId、displayOrder、favorite、locked、accessLevel等，足以支撑仪表盘清单的元数据同步场景。

五、底层请求机制：认证、限流与错误处理

1. API 基地址与请求封装

KnowiClient 使用https://knowi.com/api/1.0作为基础地址，基于requests.Session管理连接。request方法统一完成三件事：认证头注入（Authorization: Bearer ...）、Accept/Content-Type 头设置、以及响应状态码校验。非 200 响应会进入 raise_for_error 解析错误详情；5xx 状态码则直接抛出Server5xxError以触发上层重试。

2. 限流（Rate Limit）

客户端通过@utils.ratelimit(1000, 60)装饰器对check_access_token与request两个入口施加限流——即每 60 秒最多发起 1000 次请求。utils.ratelimit来自 Singer 的singer.utils，若触发限流会抛出异常并由上层按照重试策略处理。

3. 错误码到异常的类型化映射

client.py维护了一张完整的 HTTP 错误码到异常类的映射表：

状态码	异常类
400	`KnowiBadRequestError`
401	`KnowiUnauthorizedError`
402	`KnowiPaymentRequiredError`
403	`KnowiForbiddenError`
404	`KnowiNotFoundError`
405	`KnowiMethodNotAllowedError`
406	`KnowiNotAcceptableError`
408	`KnowiRequestTimeoutError`
409	`KnowiUserConflictError`
415	`KnowiUnsupportedMediaTypeError`
422	`KnowiUnprocessableEntityError`
423	`KnowiScrollExistsError`（特殊处理`scroll_exists`错误码）
500	`KnowiInternalServiceError`
其他	`KnowiError`（兜底）

这些异常类统一继承自KnowiError，其中Server5xxError、Server429Error通常会被上层认定为可重试错误，从而实现失败请求的自动重试。

4. 连接测试

Mage 在保存/测试连接时会调用 test_connection，其内部实例化客户端并执行check_access_token()。该方法请求https://knowi.com/api/1.0/dashboards这个轻量端点：只要返回 200 且响应 JSON 中包含list键，即判定令牌有效、连接成功；否则记录Error status_code = ...并抛出对应异常。这也是你在界面点"Test Connection"时背后真正执行的校验逻辑。

六、在 Mage 中接入 Knowi 的实践步骤

1. 创建数据集成管道

在 Mage 界面新建一个Data Integration类型管道；
选择数据源（Source）为Knowi（对应mage_ai/data_integrations/sources/constants.py中登记的Knowi项）；
填入上文所述的 4 个配置参数，其中access_token必填；
点击测试连接，Mage 会调用test_connection校验令牌与网络连通性；
选择要同步的 Stream（dashboards、widgets等），配置目标（Destination）即可启动抽取。

2. 命令行验证（可选）

Mage 的 Source 入口支持以脚本方式运行：

python mage_integrations/mage_integrations/sources/knowi/__init__.py --config <config.json> --state <state.json>

其中config.json的结构即 templates/config.json 中声明的字段。实际抽取由load_data驱动，每个 Stream 产生的记录以单条列表形式逐条 yield（见init.py 第 45-46 行），供下游目标按批次写入。

3. 运行与排障建议

若同步任务在开始时失败，优先检查access_token是否有效（可重新测试连接）；
若日志出现Server5xxError或 429 相关异常，说明 Knowi 侧服务异常或触发限流，可在重试窗口后恢复；
全量同步场景下start_date不参与过滤；如需时间维度增量抽取，需要对应的 Stream 支持 INCREMENTAL 复制方式。

七、总结

Knowi 数据源为 Mage 用户提供了一条将 Knowi 仪表盘与组件元数据接入数据管道的捷径。核心要点可归纳为：

认证：access_token是唯一必填项，通过 Bearer Token 完成认证，令牌失效时需要重新生成；
配置：request_timeout（默认 300 秒）、user_agent、start_date为可选参数，语义与实现一一对应；
数据流：dashboards与widgets两个可同步表采用全量复制，父 Stream 仅用于提供对象 ID 列表；
健壮性：底层具备类型化错误映射、限流与 5xx 重试机制，并有独立的连接测试入口，方便在配置阶段即验证凭证有效性。

掌握了这些细节后，你就可以在 Mage 中稳定地接入 Knowi 数据源，并将其纳入统一的数据集成与编排流程。

数据工程
数据编排
ETL
任务调度
批处理
流处理
数据集成
后端

【免费下载链接】mage-ai

🧙 Build, run, and manage data pipelines for integrating and transforming data.

项目地址：https://gitcode.com/gh_mirrors/ma/mage-ai

点击查看免费下载

在 Mage 中配置 Knowi 数据源：连接认证、配置参数与数据抽取机制详解

【免费下载链接】mage-ai

一、Knowi 数据源概览

二、配置参数详解

1.`access_token`（必填）：Knowi API 的认证凭据

2.`request_timeout`（可选）：请求超时控制

3.`user_agent`（可选）：请求标识

4.`start_date`（可选）：增量同步的起始时间点

三、如何获取`access_token`

四、可用的数据流（Streams）与复制方式

五、底层请求机制：认证、限流与错误处理

1. API 基地址与请求封装

2. 限流（Rate Limit）

3. 错误码到异常的类型化映射

4. 连接测试

六、在 Mage 中接入 Knowi 的实践步骤

1. 创建数据集成管道

2. 命令行验证（可选）

3. 运行与排障建议

七、总结

【免费下载链接】mage-ai

相关推荐

AMD量化模型生产部署终极指南：容器化、监控与性能优化全流程

OpenMetadata Salesforce 数据库连接器配置指南：认证方式、连接参数与元数据摄取全解析

OpenMetadata Alation 连接器配置指南：认证、后端数据库直连与元数据摄取详解

databasus 测试架构解析：按数据库版本复用内存容器，替代“每测试一个容器“（ADR-0013）

Redis(事务、持久化)

基于MatlabSimulin的微电网模型及光伏电池建模仿真分析

【Dify】智能生成英文友好Slug网址

Atlas 300V 24G 是运算加速卡吗？YOLO 部署实战与避坑指南

Windows下Eclipse CDT + MinGW-w64搭建C++开发环境全指南

【免费下载链接】mage-ai

一、Knowi 数据源概览

二、配置参数详解

1.access_token（必填）：Knowi API 的认证凭据

2.request_timeout（可选）：请求超时控制

3.user_agent（可选）：请求标识

4.start_date（可选）：增量同步的起始时间点

三、如何获取access_token

四、可用的数据流（Streams）与复制方式

五、底层请求机制：认证、限流与错误处理

1. API 基地址与请求封装

2. 限流（Rate Limit）

3. 错误码到异常的类型化映射

4. 连接测试

六、在 Mage 中接入 Knowi 的实践步骤

1. 创建数据集成管道

2. 命令行验证（可选）

3. 运行与排障建议

七、总结

【免费下载链接】mage-ai

相关推荐

AMD量化模型生产部署终极指南：容器化、监控与性能优化全流程

OpenMetadata Salesforce 数据库连接器配置指南：认证方式、连接参数与元数据摄取全解析

OpenMetadata Alation 连接器配置指南：认证、后端数据库直连与元数据摄取详解

databasus 测试架构解析：按数据库版本复用内存容器，替代“每测试一个容器“（ADR-0013）

Redis(事务、持久化)

基于MatlabSimulin的微电网模型及光伏电池建模仿真分析

【Dify】智能生成英文友好Slug网址

Atlas 300V 24G 是运算加速卡吗？YOLO 部署实战与避坑指南

Windows下Eclipse CDT + MinGW-w64搭建C++开发环境全指南

1.`access_token`（必填）：Knowi API 的认证凭据

2.`request_timeout`（可选）：请求超时控制

3.`user_agent`（可选）：请求标识

4.`start_date`（可选）：增量同步的起始时间点

三、如何获取`access_token`