使用 dlt 官方 Slack 验证源Verified Source构建数据管道从 OAuth 配置到增量消息加载【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dltSlack 官方dlt验证源verified source封装了 Slack API 的频道、消息、用户与访问日志读取逻辑帮助你用几行代码即可把 Slack 工作区数据加载到 duckdb、postgres、bigquery 等任意目标目的地。本文将带你走完从创建 Slack App 获取 User OAuth Token、执行dlt init slack duckdb初始化项目、配置secrets.toml凭据到运行管道并基于源码理解各资源channels、users、get_messages_resource、access_logs参数与增量加载原理的完整流程并给出可直接复制的自定义管道示例。Slack 验证源能加载哪些数据Slack 是一个面向团队与组织的流行消息与协作平台。本验证源基于“Slack API”将数据加载到你选择的目标目的地其对应的示例管道脚本为slack_pipeline.py由dlt init slack duckdb生成。该验证源暴露的 source 与 resource 如下名称描述slack检索全部 Slack 数据频道、所选频道的消息、用户、日志channels检索全部频道数据users检索全部用户信息get_messages_resource检索指定频道的全部消息access_logs检索访问日志前置准备获取 User OAuth Token要建立管道需要先在你的 Slack 工作区中创建一个 Slack App 以获得访问 Slack API 的用户令牌。进入你的 Slack 工作区点击左上角的工作区名称。选择Tools Customize Workspace。从左上角菜单选择Configure apps。点击右上角的BuildCreate a New App。选择 From scratch设置 App Name 并选择目标工作区。点击Create App确认创建。在 Features 部分导航到OAuth and Permissions。分配以下 scope名称描述admin管理工作区channels:history查看公共频道中的消息及其他内容groups:history查看私有频道中的消息及其他内容App 已被添加的频道im:history查看直接消息中的消息及其他内容App 已被添加的会话mpim:history查看群组直接消息中的消息及其他内容App 已被添加的会话channels:read查看工作区中公共频道的基本信息groups:read查看私有频道的基本信息App 已被添加的频道im:read查看直接消息的基本信息App 已被添加的会话mpim:read查看群组直接消息的基本信息App 已被添加的会话users:read查看工作区中的人员注意这些 scope 是可调整的请根据你的实际需求裁剪。在左侧 OAuth Permissions 中添加上述 scope并复制User OAuth Token。注意这里描述的 Slack UI 可能会变化官方快速入门指南见 Slack API 文档。初始化验证源dlt init slack duckdb执行以下命令完成管道脚手架搭建dlt init slack duckdb该命令将以上述 Slack 示例管道作为 source以 duckdb 作为 destination 进行初始化。若想使用其他目的地只需把duckdb替换为你要用的目的地名称即可。从实现上看dlt init命令的底层逻辑位于 dlt/_workspace/cli/_init_command.py 的init_pipeline_at_destination中它会从验证源仓库克隆并定位名为slack的验证源将源文件复制到项目目录解析示例脚本并自动替换destination参数同时把检测到的 secrets 写入secrets.toml、把依赖写入requirements.txt。因此运行成功后你的项目目录下会生成slack_pipeline.py示例管道脚本含可直接复制使用的代码片段.dlt/secrets.toml存放访问令牌等敏感信息的文件requirements.txt管道依赖清单。添加凭据在.dlt文件夹中找到secrets.toml它用于安全存储访问令牌等敏感信息请妥善保管该文件。其格式如下[sources.slack] access_token Please set me up! # please set me up!将前面复制到的 User OAuth Token 粘贴到access_token字段。最后按所选目的地文档配置对应的目的地凭据。关于凭据配置的更多说明可阅读 General Usage: Credentials。值得注意的是dlt init之所以能把令牌写入[sources.slack]这个 section是因为验证源源码在slack_source中声明了access_token: str dlt.secrets.valuedlt会按 section 结构自动匹配配置提供器config providers中的值。运行管道先安装所有必要依赖pip install -r requirements.txt运行管道python slack_pipeline.py管道运行结束后用以下命令验证数据是否正确加载dlt pipeline pipeline_name show示例管道的pipeline_name为slack你也可以使用任意自定义名称。更详细的运行指引参见 如何运行管道。源码剖析source 与 resourcesdlt基于 sources 与 resources 的概念工作。下面逐一剖析本验证源暴露的 source 与各 resource 的定义与参数。Sourceslackslacksource 从 Slack API 检索频道、所选频道的消息、用户、日志等数据from typing import Iterable from dlt.extract import DltResource from dlt.common.typing import TAnyDateTime START_DATE: pendulum.DateTime pendulum.DateTime(2024, 1, 1) dlt.source(nameslack, max_table_nesting2) def slack_source( page_size: int 100, access_token: str dlt.secrets.value, start_date: TAnyDateTime | None START_DATE, end_date: TAnyDateTime | None None, selected_channels: list[str] | None dlt.config.value, ) - Iterable[DltResource]: ...参数说明page_size每页最大条目数示例默认 100文档说明默认 1000可按需调整。access_token用于认证的 OAuth 令牌来自dlt.secrets.value即secrets.toml中的[sources.slack] access_token。start_date时间范围起点默认 2000 年 1 月 1 日。end_date时间范围终点。selected_channels要加载的频道列表未指定时默认加载全部频道。max_table_nesting2限制了嵌套字典/列表的展开深度避免 Slack 返回的复杂嵌套消息结构在目标端产生过深的表层级。Resourcechannels该函数以dltresource 形式产出全部频道数据from typing import Iterable from dlt.common.typing import TDataItem dlt.resource(namechannels, primary_keyid, write_dispositionreplace) def channels_resource() - Iterable[TDataItem]: ...primary_keyid以频道 ID 作为主键write_dispositionreplace每次加载时整体替换该表适合频道这类会全量刷新的元数据。Resourceusers该函数以dltresource 形式产出全部用户数据from typing import Iterable from dlt.common.typing import TDataItem dlt.resource(nameusers, primary_keyid, write_dispositionreplace) def users_resource() - Iterable[TDataItem]: ...与channels一样采用replace写入策略保证目标表中始终是工作区用户的完整快照。Resourceget_messages_resource该方法从 Slack API 获取指定频道的消息并为每个频道按其名称创建一个独立 resourcefrom typing import Iterable from pendulum import DateTime from dlt.common.typing import TDataItem START_DATE: pendulum.DateTime pendulum.DateTime(2024, 1, 1) END_DATE: pendulum.DateTime pendulum.DateTime(2024, 12, 31) def get_messages_resource( channel_data: dict[str, Any], created_at: dlt.sources.incremental[DateTime] dlt.sources.incremental( ts, initial_valueSTART_DATE, end_valueEND_DATE, allow_external_schedulersTrue, ), ) - Iterable[TDataItem]: ...参数说明channel_data描述具体频道的字典用于确定消息从哪里抓取。created_at可选参数利用dlt.sources.incremental定义消息检索的时间戳范围其子参数包括ts来自 Slack API 响应的时间戳字段消息时间。initial_value时间戳范围起点默认取slack_source中的start_date。end_value时间戳范围终点默认取slack_source中的end_date。allow_external_schedulers布尔值若为true允许 外部调度器 管理增量加载例如用 Airflow 做回填与增量调度。关于dlt.sources.incremental的完整用法可参考 增量加载游标指南。Resourceaccess_logs该方法从 Slack API 检索访问日志from typing import Iterable from dlt.common.typing import TDataItem dlt.resource( nameaccess_logs, selectedFalse, primary_keyuser_id, write_dispositionappend, ) # It is not an incremental resource; it just has an end_date filter. def logs_resource() - Iterable[TDataItem]: ...selectedFalse默认不加载该 resource需要时手动开启见下文自定义管道示例。primary_keyuser_id唯一标识为user_id。write_dispositionappend追加写入新数据追加到目标端已有数据之后。注意access_logs属于付费功能在管道或测试中可能无法工作——非付费账户调用会报错。同时要注意它不是增量 resource只是带end_date过滤的普通资源。自定义管道按你的方式加载 Slack 数据如果你希望创建自己的管道可以直接复用本验证源的 source 与 resource 方法。1. 配置管道指定管道名称、目的地与数据集pipeline dlt.pipeline( pipeline_nameslack, # 可使用自定义名称 destinationduckdb, # 选择合适的目的地如 duckdb、redshift、postgres dataset_nameslack_data # 可使用自定义名称 )2. 从指定起始日期加载 Slack 资源source slack_source(page_size1000, start_datedatetime.datetime(2023, 9, 1), end_datedatetime.datetime(2023, 9, 8)) # 如需只加载 access_logs取消下面这行注释仅付费账户可用 # source.access_logs.selected True # 加载 2023-09-01 至 2023-09-08 的数据。 load_info pipeline.run(source) print(load_info)后续运行将只加载自上次运行以来更新过的条目增量行为由get_messages_resource的dlt.sources.incremental游标机制保证。3. 从指定频道、指定日期范围加载数据selected_channels[general, random] # 在此填写频道名称 source slack_source( page_size20, selected_channelsselected_channels, start_datedatetime.datetime(2023, 9, 1), end_datedatetime.datetime(2023, 9, 8), ) # 从 general 和 random 两个频道加载 2023-09-01 至 2023-09-08 的数据。 load_info pipeline.run(source) print(load_info)4. 只加载所选频道的消息selected_channels[general, random] # 在此填写频道名称 source slack_source( page_size20, selected_channelsselected_channels, start_datedatetime.datetime(2023, 9, 1), end_datedatetime.datetime(2023, 9, 8), ) # 只加载 general 频道的消息。 load_info pipeline.run(source.with_resources(general)) print(load_info)source.with_resources(general)会按频道名筛选 resource由于get_messages_resource为每个频道创建一个以频道名称命名的 resource因此这里可以直接用频道名作为 resource 名进行过滤加载。小结通过本文你已掌握 Slackdlt验证源的完整使用链路创建 Slack App 并配置 OAuth scope 获取 User OAuth Token →dlt init slack duckdb初始化项目 → 在secrets.toml中注入令牌与目的地凭据 → 安装依赖并运行python slack_pipeline.py→ 用dlt pipeline slack show验证结果。同时你也理解了五个核心组件slacksource、channels、users、get_messages_resource、access_logs的源码级参数语义——尤其是get_messages_resource借助dlt.sources.incremental实现的增量加载以及with_resources按频道名精细控制加载范围的方法。基于这些资源你可以组合出完全贴合自身业务的 Slack 数据管道。延伸阅读dlt 命令行接口参考Source 与 Resource 核心概念 与 Resource 指南增量加载游标指南目标目的地一览 与 duckdb 目的地凭据配置进阶如何运行管道【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 使用 dlt 官方 Slack 验证源（Verified Source）构建数据管道：从 OAuth 配置到增量消息加载
使用 dlt 官方 Slack 验证源Verified Source构建数据管道从 OAuth 配置到增量消息加载【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dltSlack 官方dlt验证源verified source封装了 Slack API 的频道、消息、用户与访问日志读取逻辑帮助你用几行代码即可把 Slack 工作区数据加载到 duckdb、postgres、bigquery 等任意目标目的地。本文将带你走完从创建 Slack App 获取 User OAuth Token、执行dlt init slack duckdb初始化项目、配置secrets.toml凭据到运行管道并基于源码理解各资源channels、users、get_messages_resource、access_logs参数与增量加载原理的完整流程并给出可直接复制的自定义管道示例。Slack 验证源能加载哪些数据Slack 是一个面向团队与组织的流行消息与协作平台。本验证源基于“Slack API”将数据加载到你选择的目标目的地其对应的示例管道脚本为slack_pipeline.py由dlt init slack duckdb生成。该验证源暴露的 source 与 resource 如下名称描述slack检索全部 Slack 数据频道、所选频道的消息、用户、日志channels检索全部频道数据users检索全部用户信息get_messages_resource检索指定频道的全部消息access_logs检索访问日志前置准备获取 User OAuth Token要建立管道需要先在你的 Slack 工作区中创建一个 Slack App 以获得访问 Slack API 的用户令牌。进入你的 Slack 工作区点击左上角的工作区名称。选择Tools Customize Workspace。从左上角菜单选择Configure apps。点击右上角的BuildCreate a New App。选择 From scratch设置 App Name 并选择目标工作区。点击Create App确认创建。在 Features 部分导航到OAuth and Permissions。分配以下 scope名称描述admin管理工作区channels:history查看公共频道中的消息及其他内容groups:history查看私有频道中的消息及其他内容App 已被添加的频道im:history查看直接消息中的消息及其他内容App 已被添加的会话mpim:history查看群组直接消息中的消息及其他内容App 已被添加的会话channels:read查看工作区中公共频道的基本信息groups:read查看私有频道的基本信息App 已被添加的频道im:read查看直接消息的基本信息App 已被添加的会话mpim:read查看群组直接消息的基本信息App 已被添加的会话users:read查看工作区中的人员注意这些 scope 是可调整的请根据你的实际需求裁剪。在左侧 OAuth Permissions 中添加上述 scope并复制User OAuth Token。注意这里描述的 Slack UI 可能会变化官方快速入门指南见 Slack API 文档。初始化验证源dlt init slack duckdb执行以下命令完成管道脚手架搭建dlt init slack duckdb该命令将以上述 Slack 示例管道作为 source以 duckdb 作为 destination 进行初始化。若想使用其他目的地只需把duckdb替换为你要用的目的地名称即可。从实现上看dlt init命令的底层逻辑位于 dlt/_workspace/cli/_init_command.py 的init_pipeline_at_destination中它会从验证源仓库克隆并定位名为slack的验证源将源文件复制到项目目录解析示例脚本并自动替换destination参数同时把检测到的 secrets 写入secrets.toml、把依赖写入requirements.txt。因此运行成功后你的项目目录下会生成slack_pipeline.py示例管道脚本含可直接复制使用的代码片段.dlt/secrets.toml存放访问令牌等敏感信息的文件requirements.txt管道依赖清单。添加凭据在.dlt文件夹中找到secrets.toml它用于安全存储访问令牌等敏感信息请妥善保管该文件。其格式如下[sources.slack] access_token Please set me up! # please set me up!将前面复制到的 User OAuth Token 粘贴到access_token字段。最后按所选目的地文档配置对应的目的地凭据。关于凭据配置的更多说明可阅读 General Usage: Credentials。值得注意的是dlt init之所以能把令牌写入[sources.slack]这个 section是因为验证源源码在slack_source中声明了access_token: str dlt.secrets.valuedlt会按 section 结构自动匹配配置提供器config providers中的值。运行管道先安装所有必要依赖pip install -r requirements.txt运行管道python slack_pipeline.py管道运行结束后用以下命令验证数据是否正确加载dlt pipeline pipeline_name show示例管道的pipeline_name为slack你也可以使用任意自定义名称。更详细的运行指引参见 如何运行管道。源码剖析source 与 resourcesdlt基于 sources 与 resources 的概念工作。下面逐一剖析本验证源暴露的 source 与各 resource 的定义与参数。Sourceslackslacksource 从 Slack API 检索频道、所选频道的消息、用户、日志等数据from typing import Iterable from dlt.extract import DltResource from dlt.common.typing import TAnyDateTime START_DATE: pendulum.DateTime pendulum.DateTime(2024, 1, 1) dlt.source(nameslack, max_table_nesting2) def slack_source( page_size: int 100, access_token: str dlt.secrets.value, start_date: TAnyDateTime | None START_DATE, end_date: TAnyDateTime | None None, selected_channels: list[str] | None dlt.config.value, ) - Iterable[DltResource]: ...参数说明page_size每页最大条目数示例默认 100文档说明默认 1000可按需调整。access_token用于认证的 OAuth 令牌来自dlt.secrets.value即secrets.toml中的[sources.slack] access_token。start_date时间范围起点默认 2000 年 1 月 1 日。end_date时间范围终点。selected_channels要加载的频道列表未指定时默认加载全部频道。max_table_nesting2限制了嵌套字典/列表的展开深度避免 Slack 返回的复杂嵌套消息结构在目标端产生过深的表层级。Resourcechannels该函数以dltresource 形式产出全部频道数据from typing import Iterable from dlt.common.typing import TDataItem dlt.resource(namechannels, primary_keyid, write_dispositionreplace) def channels_resource() - Iterable[TDataItem]: ...primary_keyid以频道 ID 作为主键write_dispositionreplace每次加载时整体替换该表适合频道这类会全量刷新的元数据。Resourceusers该函数以dltresource 形式产出全部用户数据from typing import Iterable from dlt.common.typing import TDataItem dlt.resource(nameusers, primary_keyid, write_dispositionreplace) def users_resource() - Iterable[TDataItem]: ...与channels一样采用replace写入策略保证目标表中始终是工作区用户的完整快照。Resourceget_messages_resource该方法从 Slack API 获取指定频道的消息并为每个频道按其名称创建一个独立 resourcefrom typing import Iterable from pendulum import DateTime from dlt.common.typing import TDataItem START_DATE: pendulum.DateTime pendulum.DateTime(2024, 1, 1) END_DATE: pendulum.DateTime pendulum.DateTime(2024, 12, 31) def get_messages_resource( channel_data: dict[str, Any], created_at: dlt.sources.incremental[DateTime] dlt.sources.incremental( ts, initial_valueSTART_DATE, end_valueEND_DATE, allow_external_schedulersTrue, ), ) - Iterable[TDataItem]: ...参数说明channel_data描述具体频道的字典用于确定消息从哪里抓取。created_at可选参数利用dlt.sources.incremental定义消息检索的时间戳范围其子参数包括ts来自 Slack API 响应的时间戳字段消息时间。initial_value时间戳范围起点默认取slack_source中的start_date。end_value时间戳范围终点默认取slack_source中的end_date。allow_external_schedulers布尔值若为true允许 外部调度器 管理增量加载例如用 Airflow 做回填与增量调度。关于dlt.sources.incremental的完整用法可参考 增量加载游标指南。Resourceaccess_logs该方法从 Slack API 检索访问日志from typing import Iterable from dlt.common.typing import TDataItem dlt.resource( nameaccess_logs, selectedFalse, primary_keyuser_id, write_dispositionappend, ) # It is not an incremental resource; it just has an end_date filter. def logs_resource() - Iterable[TDataItem]: ...selectedFalse默认不加载该 resource需要时手动开启见下文自定义管道示例。primary_keyuser_id唯一标识为user_id。write_dispositionappend追加写入新数据追加到目标端已有数据之后。注意access_logs属于付费功能在管道或测试中可能无法工作——非付费账户调用会报错。同时要注意它不是增量 resource只是带end_date过滤的普通资源。自定义管道按你的方式加载 Slack 数据如果你希望创建自己的管道可以直接复用本验证源的 source 与 resource 方法。1. 配置管道指定管道名称、目的地与数据集pipeline dlt.pipeline( pipeline_nameslack, # 可使用自定义名称 destinationduckdb, # 选择合适的目的地如 duckdb、redshift、postgres dataset_nameslack_data # 可使用自定义名称 )2. 从指定起始日期加载 Slack 资源source slack_source(page_size1000, start_datedatetime.datetime(2023, 9, 1), end_datedatetime.datetime(2023, 9, 8)) # 如需只加载 access_logs取消下面这行注释仅付费账户可用 # source.access_logs.selected True # 加载 2023-09-01 至 2023-09-08 的数据。 load_info pipeline.run(source) print(load_info)后续运行将只加载自上次运行以来更新过的条目增量行为由get_messages_resource的dlt.sources.incremental游标机制保证。3. 从指定频道、指定日期范围加载数据selected_channels[general, random] # 在此填写频道名称 source slack_source( page_size20, selected_channelsselected_channels, start_datedatetime.datetime(2023, 9, 1), end_datedatetime.datetime(2023, 9, 8), ) # 从 general 和 random 两个频道加载 2023-09-01 至 2023-09-08 的数据。 load_info pipeline.run(source) print(load_info)4. 只加载所选频道的消息selected_channels[general, random] # 在此填写频道名称 source slack_source( page_size20, selected_channelsselected_channels, start_datedatetime.datetime(2023, 9, 1), end_datedatetime.datetime(2023, 9, 8), ) # 只加载 general 频道的消息。 load_info pipeline.run(source.with_resources(general)) print(load_info)source.with_resources(general)会按频道名筛选 resource由于get_messages_resource为每个频道创建一个以频道名称命名的 resource因此这里可以直接用频道名作为 resource 名进行过滤加载。小结通过本文你已掌握 Slackdlt验证源的完整使用链路创建 Slack App 并配置 OAuth scope 获取 User OAuth Token →dlt init slack duckdb初始化项目 → 在secrets.toml中注入令牌与目的地凭据 → 安装依赖并运行python slack_pipeline.py→ 用dlt pipeline slack show验证结果。同时你也理解了五个核心组件slacksource、channels、users、get_messages_resource、access_logs的源码级参数语义——尤其是get_messages_resource借助dlt.sources.incremental实现的增量加载以及with_resources按频道名精细控制加载范围的方法。基于这些资源你可以组合出完全贴合自身业务的 Slack 数据管道。延伸阅读dlt 命令行接口参考Source 与 Resource 核心概念 与 Resource 指南增量加载游标指南目标目的地一览 与 duckdb 目的地凭据配置进阶如何运行管道【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy ️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考