如何用 dlt-connector skill 把 SaaS 数据加载进 DuckDB 并自动生成 WrenAI 项目

如何用 dlt-connector skill 把 SaaS 数据加载进 DuckDB 并自动生成 WrenAI 项目

【免费下载链接】WrenAIGenBI (Generative BI) for AI agents, an open-source, governed text-to-SQL through an open context layer that turns natural-language questions into trusted dashboards, charts, and SQL across 20+ data sources, such as BigQuery, Snowflake, PostgreSQL, ClickHouse, Amazon Redshift, Databricks and more.项目地址: https://gitcode.com/GitHub_Trending/wr/WrenAI

如果你的数据存在 HubSpot、Stripe、GitHub 这类 SaaS 服务里，又希望能直接用 SQL 查询它们，WrenAI 提供了一个dlt-connectorskill：先用 dlt（data load tool）把 SaaS API 的数据抽到本地 DuckDB 文件，再自动扫描该文件生成一个可查询的 Wren 项目（YAML 模型、关系、profile），最后用真实查询验证整条链路。整条流程分四个阶段：Extract（抽取）、Model（生成项目）、Build & Verify（构建与验证）、Handoff（交付），文档强调目标是"查询真正能跑通的项目"，而不只是生成一堆看起来正确的文件。

前提：本机 Python 3.11+，并准备一个 SaaS 服务的 API 凭证。

准备条件

两条安装命令，一条装 WrenAI CLI，一条装 dlt 的 DuckDB 目标端：

pip install wrenai # core CLI（内置 DuckDB 连接器，无需额外 extra） pip install "dlt[duckdb]" --break-system-packages

如果你用的是 Claude Code 等 AI 编码代理，还可以安装一个发现 stub，之后用自然语言触发这个 skill：

npx skills add Canner/WrenAI

安装完成后新开会话（skill 在会话启动时加载），对代理说类似"connect Stripe data"的话即可路由到 dlt-connector；本文按手动操作路径完整走一遍。

dlt-connector 的参考文档和脚本都随 CLI 分发，随时可以取回：

wren skills get dlt-connector --full # SKILL.md 正文 + references/ wren skills get dlt-connector --script introspect_dlt > introspect_dlt.py

阶段一：用 dlt 把 SaaS 数据抽进 DuckDB

1. 选择数据源并安装凭证

skill 自带一份常用已验证数据源及凭证变量的速查表（dlt_sources.md）：

速查表里没有的服务，skill 要求先查 dlthub.com 的 verified sources 列表再动手，不要凭猜测拼 source 函数。

一次性运行用环境变量最省事（以 Stripe 为例，key 在 Stripe Dashboard → Developers → API keys 获取，测试可以用 test 模式 key）：

export SOURCES__STRIPE_ANALYTICS__STRIPE_SECRET_KEY="sk_test_..."

重复运行时可以改用.dlt/secrets.toml存凭证。文档提醒：凭证不要提交进 git。

2. 编写并运行 pipeline 脚本

pipeline 脚本的模式固定为：配置destination="duckdb"，然后pipeline.run(source)。以 Stripe 为例（来自 skill 的速查表模板）：

import dlt from stripe_analytics import stripe_source pipeline = dlt.pipeline( pipeline_name="stripe", destination="duckdb", dataset_name="stripe_data", ) source = stripe_source() info = pipeline.run(source) print(info)

其他数据源（HubSpot、GitHub、Slack、Salesforce 等）按同一模式替换 pipeline 名和 source 函数即可，速查表里给出了各自的模板和典型产出表，例如 HubSpot 常见产出 contacts、companies、deals 等表。

运行方式：

python <你的pipeline脚本>.py

3. 确认数据已落盘

运行结束后文档要求确认三件事：pipeline 没有报错；生成了.duckdb文件（一般是<pipeline_name>.duckdb，本例即stripe.duckdb）；打印发现的表。skill 给了一段检查脚本，把其中的文件名换成你的即可：

import duckdb con = duckdb.connect("stripe.duckdb", read_only=True) for row in con.execute(""" SELECT table_schema, table_name, (SELECT COUNT(*) FROM information_schema.columns c WHERE c.table_schema = t.table_schema AND c.table_name = t.table_name) as col_count FROM information_schema.tables t WHERE table_schema NOT IN ('information_schema', 'pg_catalog') AND table_name NOT LIKE '_dlt_%' ORDER BY table_schema, table_name """).fetchall(): print(f" {row[0]}.{row[1]} ({row[2]} columns)") con.close()

能列出业务表（而非只有_dlt_%内部表）说明抽取成功。如果你手头已有之前 dlt 跑出来的.duckdb文件，可以直接从阶段二开始。

阶段二：自动生成 Wren 项目

1. 两个硬性约定

skill 把下面两条列为不可协商的规则，理解了它们才好判断生成的项目对不对：

1. DuckDB catalog 命名：Wren 引擎ATTACH一个.duckdb文件时，用文件名的 stem（去掉扩展名的部分）作为 catalog 别名。因此每个模型table_reference.catalog必须等于文件名 stem——stripe.duckdb对应 catalogstripe。写错会在查询时报 "table not found"。
2. 类型归一化：列类型必须经过 Wren SDK 的wren.type_mapping.parse_type()（基于 sqlglot）转换，DuckDB 特有的HUGEINT、TIMESTAMP WITH TIME ZONE等需要转成规范 SQL 类型，不能手写映射表。

两条规则introspect_dlt.py脚本都会自动处理，不需要也不应该手动覆盖。

2. 运行 introspect_dlt 脚本

wren skills get dlt-connector --script introspect_dlt > introspect_dlt.py python introspect_dlt.py \ --duckdb-path ./stripe.duckdb \ --output-dir ./project \ --project-name stripe

参数说明：--duckdb-path指阶段一产出的文件；--output-dir是项目落盘目录；--project-name缺省时取自文件名。脚本只读地连接 DuckDB，完成以下工作（对应脚本 introspect_dlt.py 的实现）：

• 通过information_schema发现所有用户表，过滤掉_dlt_loads、_dlt_pipeline_state等内部表；
• 把_dlt_id、_dlt_parent_id、_dlt_load_id、_dlt_list_idx等元数据列从模型定义中隐藏；
• 依据_dlt_parent_id列和表命名约定（子表名 = 父表名__后缀）检测父子关系；
• 生成完整项目：wren_project.yml（data_source: duckdb）、models/<表名>/metadata.yml、relationships.yml、knowledge/。

如果目标目录已有wren_project.yml，脚本会报错退出，需要加--force覆盖。

运行输出的示例（文档中脚本的实际打印格式）：

Introspecting .../stripe.duckdb... Catalog (from filename): stripe Found 11 tables Detected 3 parent-child relationships Wren project written to .../project/ 11 models, 3 relationships

脚本建议的后续动作就是wren context validate和wren context build，见下一阶段。

3. 抽查生成的模型

skill 要求抽查至少一个模型，确认四件事：

1. table_reference.catalog与 DuckDB 文件名一致（stripe.duckdb→stripe）；
2. table_reference.schema与 DuckDB 中的 schema 一致（一般是main）；
3. columns 列表里没有出现_dlt_*列；
4. 列类型是规范 SQL 类型（VARCHAR、BIGINT、BOOLEAN、TIMESTAMP 等）。

4. 配置指向该 DuckDB 文件的连接 profile

profile 存在~/.wren/profiles.yml，之后所有 CLI 命令都通过它连接数据库。注意url必须填.duckdb文件所在的目录，而不是文件本身。skill 给出的配置脚本（其中 profile 名、duckdb 路径按需替换）：

import yaml from pathlib import Path wren_home = Path.home() / ".wren" wren_home.mkdir(exist_ok=True) profiles_file = wren_home / "profiles.yml" existing = ( (yaml.safe_load(profiles_file.read_text()) or {}) if profiles_file.exists() else {} ) existing.setdefault("profiles", {}) profile_name = "stripe_dlt" existing["profiles"][profile_name] = { "datasource": "duckdb", "url": str(Path("./stripe.duckdb").resolve().parent), "format": "duckdb", } existing["active"] = profile_name profiles_file.write_text(yaml.dump(existing, default_flow_style=False, sort_keys=False))

配置完成后可以用wren profile list确认该 profile 处于 active 状态；也可以进入项目目录执行wren context set-profile <profile名>，把它写进wren_project.yml，之后项目内的命令固定使用该 profile。

阶段三：构建并用真实查询验证

这一阶段文档明确说"不可跳过"——生成了 YAML 但查询跑不通不算成功。

1. 构建 MDL

cd project wren context build

这一步把 YAML 模型编译成target/mdl.json。失败时先按下一节的排查项处理。构建通过后，dlt-connector skill 的四阶段流程还会执行wren memory index完成记忆索引，然后进入验证。

2. 对每个模型跑查询

至少对每个模型各跑一条查询，确认能解析并返回数据：

wren --sql 'SELECT COUNT(*) as total FROM "customers"'

再挑 2–3 条有代表性的查询让用户看到数据（表名换成实际表名）：

wren --sql 'SELECT * FROM "invoices" LIMIT 5'

如果存在父子关系，验证父表、子表两侧都能查询。查询返回真实数据之后，这个 Wren 项目才算完成。

排查与限制

skill 给出的排查清单（按症状对应）：

• wren context build失败：检查wren_project.yml里data_source: duckdb是否设置；确认 profile 中 DuckDB 文件路径正确；用wren context validate获取详细报错。
• 查询报 "table not found"：最常见原因是table_reference.catalog与 DuckDB 文件名不一致（pipeline.duckdb对应 catalog 必须是pipeline，不能是空串）；同时检查 profile 的url是否指向文件所在目录；含双下划线的表名要加引号，如"hubspot__contacts"。
• 类型报错：检查模型 YAML 中列类型是否为规范 SQL 类型；在已安装 wren SDK 的环境里重跑introspect_dlt.py重新做类型归一化。
• DuckDB 文件被锁：DuckDB 是单写者模型，dlt pipeline 运行期间不要并发查询，等它结束；需要并发写入时应让 dlt 写另一个文件再原子替换。

另一个已知边界：dlt 的_dlt_parent_id/_dlt_id列仍保留在 DuckDB 表里（关系条件要用它们），只是不出现在 Wren 模型定义中；生成的模型一律使用table_reference而不是ref_sql，因为它们直接映射 dlt 创建的 DuckDB 表。

下一步

项目是 DuckDB 支撑的，这正是 WrenAI GenBI 快照模式要的数据形态。如果要把这批数据做成可分享的浏览器端 dashboard 并部署到 Vercel / Cloudflare Pages，交接给 genbi skill：wren skills get genbi，它的快照源就是这条 pipeline 产出的.duckdb文件。

相关文档：Skills 参考、dlt-connector SKILL.md、连接你的数据库。

【免费下载链接】WrenAIGenBI (Generative BI) for AI agents, an open-source, governed text-to-SQL through an open context layer that turns natural-language questions into trusted dashboards, charts, and SQL across 20+ data sources, such as BigQuery, Snowflake, PostgreSQL, ClickHouse, Amazon Redshift, Databricks and more.项目地址: https://gitcode.com/GitHub_Trending/wr/WrenAI