wren-dlt-connector

wren-dlt-connector

将 SaaS 数据连接到 Wren Engine，以便进行 SQL 分析——通过一次对话，从零开始构建并验证可查询的项目。

适用对象

懂 SQL 和一些 Python 的数据分析师，但可能之前没用过 dlt 或 Wren。首次出现概念时，简要说明即可，不必过度解释 SQL 熟手本来就了解的内容。

概述

此技能将引导用户完成包含四个阶段的工作流：

1. 提取 — 使用 dlt（data load tool）将数据从 SaaS API 提取到本地 DuckDB 文件
2. 建模 — 检查 DuckDB 架构并自动生成 Wren 语义项目（YAML 模型、关系、配置文件）
3. 构建并验证 — 构建项目并运行实际 SQL 查询，确认端到端流程正常
4. 交接 — 向用户展示其数据和后续步骤

用户可能从任意阶段开始。询问他们要从哪个阶段开始——他们可能已经有一个 .duckdb 文件，只需要完成阶段 2–4。

目标是构建一个实际能够成功查询的项目，而不只是生成看起来正确的文件。 宣告成功前，务必执行验证步骤。

关键事项：DuckDB 目录命名

当 wren engine 连接到 DuckDB 文件时，会使用文件名（不含 .duckdb 扩展名）作为目录别名执行 ATTACH：

ATTACH DATABASE 'stripe_data.duckdb' AS "stripe_data" (READ_ONLY)

这意味着每个模型的 table_reference.catalog 都必须与 DuckDB 文件名主体相同。如果文件名是 hubspot.duckdb，目录就是 hubspot。如果文件名是 my_pipeline.duckdb，目录就是 my_pipeline。

如果设置错误，查询时会出现“找不到表”错误。introspect_dlt.py 脚本会自动处理这一点。

关键事项：类型规范化

列类型必须使用 wren SDK 的 type_mapping.parse_type() 函数进行规范化。该函数使用 sqlglot 将数据库特定类型（例如 DuckDB 的 HUGEINT、TIMESTAMP WITH TIME ZONE）转换为 wren-core 能识别的规范 SQL 类型。不要硬编码类型映射——始终交由 parse_type(raw_type, "duckdb") 处理。

安装 wren SDK 后，introspect_dlt.py 脚本会自动完成此操作。

阶段 1：提取 — 设置 dlt 管道

步骤 1：选择 SaaS 数据源

询问用户要连接哪个 SaaS 服务。阅读 references/dlt_sources.md，了解常用的已验证数据源及其身份验证要求。如果列表中没有该数据源，搜索 dlthub.com/docs/dlt-ecosystem/verified-sources，确认 dlt 是否为其提供了已验证的数据源。

步骤 2：安装 dlt

pip install "dlt[duckdb]" --break-system-packages

步骤 3：编写管道脚本

创建一个 Python 脚本，以便：

1. 导入所选 SaaS 的 dlt 数据源函数
2. 使用 destination='duckdb' 和本地文件路径配置管道
3. 使用 pipeline.run(source) 运行管道

以下是通用模式——请根据具体数据源进行调整（各数据源专用模板请参阅 references/dlt_sources.md）：

import dlt

pipeline = dlt.pipeline(
    pipeline_name="<source>_pipeline",
    destination="duckdb",
    dataset_name="<source>_data",
)

# Source-specific: check references/dlt_sources.md for auth patterns
source = <source_function>(api_key=dlt.secrets.value)

info = pipeline.run(source)
print(info)

步骤 4：设置凭据

dlt 会从环境变量或 .dlt/secrets.toml 读取凭据。对于一次性运行，最简单的做法是：

# Set the credential as an environment variable
# The exact variable name depends on the source — check references/dlt_sources.md
export SOURCES__<SOURCE>__API_KEY="the-actual-key"

询问用户的 API 密钥或令牌。提醒他们：

• 切勿将凭据提交到 git
• 一次性运行时，使用环境变量最简单
• 如需重复使用，可以创建 .dlt/secrets.toml

步骤 5：运行管道

python <pipeline_script>.py

运行后，确认：

1. 管道已成功完成，且没有错误
2. 已创建 .duckdb 文件（通常为 <pipeline_name>.duckdb）
3. 打印发现的表及其列数

import duckdb
con = duckdb.connect("<pipeline_name>.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()

阶段 2：模型 — 生成 Wren 项目

运行内省脚本，从 DuckDB 文件自动生成完整的 Wren 项目：

python <path-to-this-skill>/scripts/introspect_dlt.py \
    --duckdb-path <path-to-duckdb-file> \
    --output-dir <project-directory> \
    --project-name <name>

此脚本将：

• 以只读方式连接 DuckDB 文件
• 将 table_reference.catalog 设置为 DuckDB 文件名（不含扩展名）（与 wren 引擎的 ATTACH 行为一致）
• 通过 information_schema 查找所有表和列
• 过滤掉 dlt 内部表（_dlt_loads、_dlt_pipeline_state 等）
• 从模型定义中过滤掉 dlt 元数据列（_dlt_id、_dlt_load_id、_dlt_list_idx）
• 根据 _dlt_parent_id 列和表命名约定检测父子关系
• 使用 wren.type_mapping.parse_type() 规范化列类型（基于 sqlglot）
• 生成完整的 v2 YAML 项目（wren_project.yml、models/、relationships.yml、instructions.md）

运行后，向用户展示生成的内容：

# Show project summary
cat <project-directory>/wren_project.yml
echo "---"
ls <project-directory>/models/
echo "---"
cat <project-directory>/relationships.yml

验证模型是否正确

抽查一个生成的模型，确认：

1. table_reference.catalog 与 DuckDB 文件名一致（例如，stripe_data.duckdb 对应 stripe_data）
2. table_reference.schema 与 DuckDB 架构一致（通常为 main）
3. 列表中没有 _dlt_* 列
4. 列类型看起来合理（VARCHAR、BIGINT、BOOLEAN、TIMESTAMP 等）

设置连接配置

创建一个 Wren 配置文件，让用户无需每次都指定连接详情即可查询。url 必须指向包含 .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 = "<source>_dlt"
existing["profiles"][profile_name] = {
    "datasource": "duckdb",
    "url": str(Path("<duckdb-path>").resolve().parent),
    "format": "duckdb",
}
existing["active"] = profile_name

profiles_file.write_text(yaml.dump(existing, default_flow_style=False, sort_keys=False))

阶段 3：构建并验证 — 项目必须实际可用

此阶段不可跳过。能生成 YAML、却在查询时失败的项目并不算成功。

步骤 1：构建 MDL

cd <project-directory>
wren context build

此操作会将 YAML 模型编译为 target/mdl.json。如果失败，请先修复问题再继续（参见下文的“故障排除”）。

步骤 2：使用实际查询进行验证

至少对每个生成的模型运行一次查询，以确认项目可正常使用：

# For each model, verify it resolves correctly
wren --sql 'SELECT COUNT(*) as total FROM "<table_name>"'

如果任何查询失败，请先调试并修复模型，再继续。常见问题：

• table_reference 中的 catalog 错误 → “未找到表”
• 类型不匹配 → 修复 metadata.yml 中的列类型
• 缺少配置文件 → 检查 wren profile list

步骤 3：运行有意思的查询

基本查询通过后，再运行 2–3 个更有意思的查询，向用户展示其数据的样子：

# Preview data
wren --sql 'SELECT * FROM "<table_name>" LIMIT 5'

# If there's a relationship, verify both models are queryable
wren --sql 'SELECT * FROM "<parent>" LIMIT 5'
wren --sql 'SELECT * FROM "<child>" LIMIT 5'

向用户展示结果，并解释他们看到的内容。这是用户首次通过 Wren 查看数据，务必让这次体验有所收获。

步骤 4：确认成功

只有在查询返回实际数据后，才能告知用户设置已完成。总结以下内容：

• 创建了多少个模型
• 检测到了哪些关系
• 当前激活的是哪个配置文件
• 用户接下来可以尝试的示例查询

故障排除

如果 wren context build 失败：

• 检查 wren_project.yml 中是否设置了 data_source: duckdb
• 确认配置文件中的 DuckDB 文件路径正确
• 运行 wren context validate 查看详细错误信息

如果查询因“未找到表”而失败：

• 最可能的原因： table_reference.catalog 与 DuckDB 文件名不一致。如果文件名为 pipeline.duckdb，catalog 必须是 pipeline，不能是空字符串。
• 检查配置文件的 url 是否指向包含 .duckdb 文件的目录
• 包含双下划线的表名需要加引号："hubspot__contacts"

如果查询因类型错误而失败：

• 检查模型 YAML 中的列类型——它们应为规范的 SQL 类型（VARCHAR、BIGINT 等）
• 安装 wren SDK 后重新运行 introspect_dlt.py，以正确规范化类型

常规检查：

• 检查配置文件是否处于激活状态：wren profile list
• 如果 dlt 管道正在运行，DuckDB 文件可能被锁定——请等待运行结束

重要说明

• dlt 的 _dlt_parent_id / _dlt_id 列会保留在实际的 DuckDB 表中，但在 Wren 模型定义中隐藏。这些列仅用于关系条件。
• DuckDB 有单写入者限制。查询期间不要运行 dlt 同步。若要支持并发访问，dlt 应写入单独的文件，然后以原子方式替换。
• 生成的模型使用 table_reference（而非 ref_sql），因为它们直接映射到 dlt 创建的 DuckDB 表。
• 列类型使用 wren SDK 的 parse_type() 和 sqlglot 的 DuckDB 方言进行规范化。如果类型看起来不正确，用户可以直接编辑模型的 metadata.yml。