implementing-warehouse-sources

Name: implementing-warehouse-sources
Author: posthog

github.com/posthog数据与数据库40.5k186 次安装今天更新

代码仓库

关于 implementing-warehouse-sources

面向数据工程师的仓库技能，指导构建或更新数据仓库数据源，涵盖数据源选择、实现模式及项目检查。

主要功能

01选择数据源基类
02共享 REST 框架指南
03可恢复与 Webhook 模式
04维护数据源清单

适用场景

01为厂商 API 添加数据源
02更新现有仓库数据源

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/posthog/posthog --skill implementing-warehouse-sources

安装后在 Claude Code 中输入/implementing-warehouse-sources确认已加载。

附带文件references/oauth-app-deployment.mdreferences/testing.md

详细介绍

实现数据仓库数据源

在 products/warehouse_sources/backend/temporal/data_imports/sources/ 中构建或更新数据仓库数据源时，使用此 skill。

先读这些

在开始编码之前，请阅读：

products/warehouse_sources/backend/temporal/data_imports/sources/source.template（文件顶部的 TODO 就是引导清单；但仍要对照当前的数据源实现来核实目标文件，因为模板可能会过时）
products/warehouse_sources/backend/temporal/data_imports/sources/README.md
products/warehouse_sources/backend/temporal/data_imports/sources/SOURCES.md — 列出了每个已注册数据源及其通信方式（HTTP / 供应商 SDK / gRPC / 数据库协议 / webhook）和 tracked-transport 状态。先略读此文件，了解类似数据源是如何接线的，以及你正在处理的数据源当前处于什么状态。请保持其同步 — 见下文“更新 SOURCES.md”。
products/warehouse_sources/backend/temporal/data_imports/sources/common/base.py — 基类（SimpleSource、ResumableSource、WebhookSource）以及 FieldType 联合类型
products/warehouse_sources/backend/temporal/data_imports/sources/common/resumable.py — ResumableSourceManager
products/warehouse_sources/backend/temporal/data_imports/sources/common/webhook_s3.py — WebhookSourceManager
chargebee/ — 新增 REST 数据源的规范参考。 它使用共享的 rest_source 框架（声明式 RESTAPIConfig + rest_api_resource、框架自带的认证与分页器、tracked+retrying transport），并且是可恢复的 — 这证明该框架覆盖了最主要的“对列表端点分页并产出，且可恢复”形态。请先读它，同时阅读下文“优先使用共享 REST 框架”。仅在需要 定制传输 兜底方案时，才阅读 klaviyo/ 或 github/：它们为大多数数据源都不具备的边缘情况（自定义查询字符串编码、多级扇出、JSON:API 重塑）手写了客户端 — 不要把那些样板代码复制到不需要它的数据源中。对于依赖资源扇出（使用 type: "resolve" 的父→子），还需阅读 products/warehouse_sources/backend/temporal/data_imports/sources/common/rest_source/__init__.py 和 config_setup.py（例如 process_parent_data_item、make_parent_key_name）。
对于支持 webhook 的数据源，请阅读 products/warehouse_sources/backend/temporal/data_imports/sources/stripe/source.py 作为参考实现。

选择合适的基础类

每个新的 source 必须继承以下之一（或它们的组合）：

SimpleSource[Config] — 适用于简单的基于拉取的 API 的默认选择，每次运行都会完整遍历端点。
ResumableSource[Config, ResumableData] — 对于任何基于 API 的新 source，只要其底层 API 支持断点续传（游标/Link 头分页、时间窗口、偏移令牌，或任何其他能够确定性地从上次中断处继续的方式），都优先使用它。如果 API 提供了下一页令牌、Link 头或稳定的时间过滤器，就使用 ResumableSource。这样 Temporal 就能在心跳超时后恢复，而无需从头重新开始。save_state 将游标暂存在内存中；流水线在写入目前已产出的行之后，立即将其提交到 Redis（24h TTL）。
WebhookSource[Config] — 仅当 source 能向我们推送事件时使用（例如 Stripe webhook 端点）。通常与 ResumableSource 组合使用，这样初始回填可断点续传，后续增量则通过 webhook 传入。

当两者都适用时，通过多重继承进行组合，例如：

class StripeSource(
    ResumableSource[StripeSourceConfig, StripeResumeConfig],
    WebhookSource[StripeSourceConfig],
    OAuthMixin,
):
    ...

经验法则：

仅拉取 API，没有可持久化的游标 → SimpleSource。
仅拉取 API，但有任何可在两次运行之间保存的游标/下一页/时间过滤器 → ResumableSource。
Source 能通过变更事件回调我们 → 在适合的拉取基础类之上再加 WebhookSource。
Source 的下一次运行从一个只有该 source 才能计算的位置开始，而非某一列的最大值 → 添加 CursorSource。

数据库和文件传输 source（SFTP、S3）继续使用 SimpleSource，除非有明确的理由另作选择。

持久化 source 游标（`CursorSource`）

增量字段覆盖了常见情况：流水线取某一列的最大值，下一次运行则过滤大于它的值。有些位置并不是某一列的最大值：例如读取前捕获的 Postgres xmin 上限，或一组 Kafka 分区偏移量。对于这些情况，应添加来自 sources/common/cursor.py 的 CursorSource[CursorT] mixin，而不是在 SourceResponse 上新增字段或在 sync_type_config 中新增键。

将游标定义为一个带有 cursor_kind: ClassVar[str] 的 @frozen dataclass。保持 kind 稳定，因为 kind 一旦改变，所有已存储的游标都会被丢弃。为后续新增的字段提供默认值，因为缺少必需字段的已存储游标会被丢弃。
实现 cursor_class()。当某次运行读取的数据较少时，若游标不能回退，则重写 merge_cursors(current, candidate)（Kafka 保留各分区的最大值）。仅当迁移以其他键存储的状态时，才重写 cursor_from_legacy()。
在 source_for_pipeline 中，调用 self.get_cursor_manager(inputs)。load() 返回已存储的游标，在重置或重建表时返回 None。stage(cursor) 将下一个游标交给流水线。
流水线仅在本次运行的行已持久化之后才持久化暂存的游标（在 v3 上，loader 会随最终批次一起提升它），而重置会清除它。postgres/xmin_cursor.py 是参考实现。

优先使用共享 REST 框架

大多数 REST 数据源都应基于共享的 rest_source 框架（common/rest_source/）构建，而不是手写客户端。它已经提供了以下能力——你完全不需要自己写这些：

带跟踪、可重试的传输层——RESTClient 默认使用 make_tracked_session()，并会重试 429 + 瞬时 5xx，同时遵循 Retry-After。无需 tenacity，无需 RetryableError，无需 fetch 循环。
分页器（rest_source/paginators.py，在配置中通过字符串/字典选择，无需手写）： single_page、header_link、json_response（响应体中的 next-URL）、cursor、offset、page_number。
认证（rest_source/auth.py）：bearer、api_key（header/query/cookie）、http_basic、oauth2 （客户自有的 client-credentials/refresh）。每种方式都会自行脱敏各自的密钥——无需 _get_headers 构造器。
增量参数、data_selector、响应操作、断点续传（resume_hook / initial_paginator_state），以及父/子扇出（fanout.build_dependent_resource）。

chargebee/ 是典型示例（声明式端点 + 框架认证 + 断点续传）。zendesk/ 展示了多端点 + data_selector；attio/ 展示了游标分页。

什么时候手写才合理（先读 klaviyo/）：API 需要框架无法生成的查询字符串（字面量的方括号/运算符，例如 filter=greater-than(...)、page[size]）；多层级（2 层以上）扇出；或逐项重塑是 data_selector 无法表达的（例如把 JSON:API 的 attributes 展平到行根）。单层级扇出和逐项映射都是以声明式方式支持的——不要为这些手写。如果你必须手写，仍然要沿用 make_tracked_session()，并且不要再加第二层状态码重试（见“重试与限流”）。

新 API 数据源的端到端工作流

按此顺序操作。每一步都对应 source.template 中的 TODO。

调研数据源。 选择用户真正会用到的端点。交叉参考：
- Airbyte：https://airbyte.com/connectors（连接器页面通常会链接到源代码——是有用的参考）
- Fivetran：https://www.fivetran.com/connectors
- Stitch：https://www.stitchdata.com/docs/integrations/ 找到官方 API 文档或 OpenAPI 规范，并在编写任何请求代码之前，确定厂商最新正式可用（GA）的 API 版本——数据源必须基于该版本构建。查看厂商的更新日志、版本或弃用页面，而不只是排名最靠前的那个页面；文档站点通常默认显示较旧的版本，而且 Airbyte/Fivetran 连接器往往落后数年。关于什么算最新，以及当最新通道尚未 GA 时该怎么做，见“厂商 API 版本元数据”。
引导数据源。 复制模板并接好枚举/类型引用：
```
mkdir -p products/warehouse_sources/backend/temporal/data_imports/sources/{SOURCE_NAME}
cp products/warehouse_sources/backend/temporal/data_imports/sources/source.template products/warehouse_sources/backend/temporal/data_imports/sources/{SOURCE_NAME}/source.py
```
然后添加那一个枚举成员。位于 products/warehouse_sources/backend/types.py 的 ExternalDataSourceType（从 products/warehouse_sources/backend/facade/types.py 重新导出）是唯一声明数据源类型的地方——遵循该文件中现有的约定：ALL_CAPS 且单词之间不带下划线 （例如 ACTIVECAMPAIGN、APPLESEARCHADS），值为 PascalCase。（整个流程中唯一的 kebab-case 标识符是可选的 featureFlag="dwh-{source_name}"。）
选择基类（见上文），并重命名类／source_type 返回值。
定义 get_source_config——名称、类别（必填——见“源类别和关键词”）、标签、说明文字、docsUrl、iconPath、字段以及可选的 keywords。使用合适的字段类型（见下文）。同时设置供应商 API 版本元数据类属性——见“供应商 API 版本元数据”。源发布时，还要按“代码库检测”中的说明设置 codebase_detection。
注册源——在 products/warehouse_sources/backend/temporal/data_imports/sources/__init__.py 中添加导入行，并将其加入 __all__。（类上的 @SourceRegistry.register 装饰器会处理运行时注册。）
运行配置生成器：pnpm run generate:source-configs。确认新的配置类出现在 products/warehouse_sources/backend/temporal/data_imports/sources/generated_configs/<your_source>.py 中（每个源对应一个生成的模块；包的 __init__.py 是手写的，不会重新生成）。不要手动编辑生成的模块。 每次更改 get_source_config.fields 后，都要重新运行生成器。
在 source.py 中，将通用的 Config 类型替换为生成的 {Source}SourceConfig 类。
实现：validate_credentials、get_schemas、source_for_pipeline（以及按需实现 get_resumable_source_manager／get_webhook_source_manager）。
拆分传输逻辑。 将 API 客户端、分页器、行规范化和 SourceResponse 组装放在 {source}.py 中。将端点目录／增量字段／主键／分区默认值放在 settings.py 中。
添加图标。 放在 frontend/public/services/{source}.png——PNG 是仓库惯例（约 800 个 PNG 对 58 个 SVG，且 source.template 默认使用 .png）。SVG 也可接受，但并非常用格式；将 iconPath 设为与你提交的扩展名一致。如果尚未提交该徽标，请从 Logo.dev 获取——向用户索要 Logo.dev API 密钥；不要将密钥硬编码。Logo.dev 的图像 API 返回 PNG（而非 SVG）。请控制文件大小。
运行迁移。 DEBUG=1 python manage.py makemigrations && DEBUG=1 ./bin/migrate（仅当新增枚举值触发 Django 迁移时才需要）。
重新生成前端 API 类型：hogli build:openapi。此命令会根据 source_config.py 中的源配置契约重新生成 products/warehouse_sources/frontend/generated/api.schemas.ts，并使源出现在前端下拉列表中。每当 source_config.py 或源的字段发生更改时，都要重新运行。
发布状态——完成的源不带 unreleasedSource 标志。 此技能产出的交付物默认不带 unreleasedSource——完成且可正常工作的源应当以可见、可连接的状态发布。无需任何人批准即可将其作为已发布源上线；这就是完成状态。搭建的存根预先设置了 unreleasedSource=True，因此完成源时需要删除该行——请直接移除。（原因：unreleasedSource=True 会让用户完全看不到连接器——前端会筛掉所有该值为真的源；参见 InlineSourceSetup.tsx 和 nonHogFunctionTemplatesLogic.tsx 中的“即将推出／通知我”流程。） 删除该行是强制要求，且不取决于任何你在当前环境中做不到的事情。 尤其是，"我无法 curl 线上 API"或"我无法针对真实账户进行验证"并不是保留该标志的理由——这正是 releaseStatus=ReleaseStatus.ALPHA 的用途（对_可见_来源打上的一个"新上线、测试较少"的软标签）。unreleasedSource=True 唯一正当保留的情形是：该来源客观上还无法同步，因为它正通过多个 PR 分阶段合入，且实现代码尚未全部到位。一个具备可用的 get_schemas / source_for_pipeline 且测试通过的来源就是已完成状态——该标志应当移除。绝不要编写断言 unreleasedSource is True 的测试——那会把 bug 锁死，正是它导致 166 个已完成的来源被隐藏，直到被迫批量发布。

因此，一个新完成、经过测试的来源应以如下形式发布：
- 不含 unreleasedSource（可见且可连接），
- 对尚未经过充分测试的新来源使用 releaseStatus=ReleaseStatus.ALPHA（待粗糙之处打磨完成后改用 ReleaseStatus.BETA；正式可用时使用 ReleaseStatus.GA，或完全省略 releaseStatus）——这是对_可见_来源打的软标签，而非门槛，
- 可选的 featureFlag="dwh-{source_name}"（kebab-case）仅当你希望面向被标记的用户进行受控发布，而非向所有人发布时才使用。
每当你设置 releaseStatus 时，请使用来自 products.warehouse_sources.backend.facade.source_config 的 ReleaseStatus 枚举——绝不要使用裸字符串字面量。请将 ReleaseStatus 加入你从该模块导入的现有 import 块中。
为来源编写文档。 遵循 /documenting-warehouse-sources 技能（模板、共享片段、<SourceParameters /> + <SourceTables />），在 posthog.com 上编写或更新面向用户的文档。确保 get_source_config 中的 docsUrl 与文档文件名（kebab-case）一致，并且——如果 get_schemas 是静态端点目录——设置 lists_tables_without_credentials = True（见下文），以便文档的 Supported tables 部分能够渲染。一个已完成的来源应附带一致的文档，而不是占位内容。
在提交 PR 前删除模板 TODO 注释。

来源架构约定

对于基于 API 的来源，请采用以下划分：

source.py：来源注册、来源表单字段、schema 列表、凭据校验、可恢复/webhook 管理器接线、管道交接。
settings.py：端点目录、增量字段、主键、分区默认值。
{source}.py：API 客户端/鉴权、分页器、请求参数、行规范化以及 SourceResponse。

这样可让端点行为保持声明式且易于扩展。

源的行为放在源中，永远不要放在 API 层

warehouse_sources 表现层（products/warehouse_sources/backend/presentation/views/external_data_source.py、external_data_schema.py）必须保持与源无关。不要在那里添加 if source_type == ExternalDataSourceType.X / source.is_direct_<engine> 分支 —— CI 守卫（.github/scripts/check-dwh-source-agnostic.py）会拦截新增的分支。

当某个源需要 API 必须调用的行为时，应改为在该源上暴露它：

API 读取的布尔值/值 → 在 _BaseSource 上添加一个带安全默认值的标志（如 supports_column_selection、connection_host_fields、has_managed_hogql_schema），让 API 基于该标志进行分支。
只有部分源才有的方法（CDC、xmin、webhook、自定义清单） → 一个由源选择加入的能力 mixin；API 通过 isinstance(source, <Capability>) 进行分派。
直接查询引擎行为（SQL 引擎如何解析表位置、构建其 DataWarehouseTable、映射列）以引擎为键，而不是以源类型为键 —— 通过引擎适配器/注册表基于 source.direct_engine 分派（查询相关关注点用 posthog/hogql/direct_sql/，物化用 data_warehouse 引擎注册表），永远不要用 source_type。

将源领域的语义（如何与引擎通信、如何命名、过滤器是否下推）保留在源上；它驱动的仓库领域工作（DataWarehouseTable 行、托管 viewset、hog function）保留在 data_warehouse 中，以源或适配器返回的内容为键。源能力永远不要导入 data_warehouse 类型。参见 products/data_warehouse/backend/presentation/README.md。

对于混合了顶层端点和扇出端点的 REST 源，将端点元数据保留在 settings.py 中，并在 {source}.py 中按以下优先级进行路由：

端点特定的自定义迭代器（仅在必需时），
通用扇出辅助路径，
顶层端点路径。

规范描述（语义增强）

表同步后，一个后台 activity（workflow_activities/enrich_table_semantics.py）会写入 WarehouseColumnAnnotation 行，描述每个表/列，并呈现给 AI Agent。对于固定 schema 的源（SaaS API）， schema 对所有人都是相同的，因此应从官方 API 文档将其一次性记录，而不是花钱让大模型为每个团队重新推导。这些精选的描述是权威的 —— 它们会被直接应用（description_source="canonical"），并且永远不会发送给大模型。

添加一个与源相伴的 canonical_descriptions.py（与 source.py / settings.py 同级）：

# products/warehouse_sources/backend/temporal/data_imports/sources/{source}/canonical_descriptions.py
from products.warehouse_sources.backend.temporal.data_imports.sources.common.canonical_descriptions import CanonicalDescriptions

CANONICAL_DESCRIPTIONS: CanonicalDescriptions = {
    "Charge": {  # key = ExternalDataSchema.name (the endpoint name from get_schemas / ENDPOINTS)
        "description": "A single attempt to move money into your account by charging a payment source.",
        "docs_url": "https://stripe.com/docs/api/charges",  # passed to the LLM for columns not covered here
        "columns": {  # column name -> one-line description, taken from the official API docs
            "id": "Unique identifier for the charge.",
            "amount": "Amount intended to be collected, in the smallest currency unit (e.g. cents).",
        },
    },
}

然后在源类上重写该 hook，并对同级文件进行惰性导入：

def get_canonical_descriptions(self) -> CanonicalDescriptions:
    from products.warehouse_sources.backend.temporal.data_imports.sources.{source}.canonical_descriptions import CANONICAL_DESCRIPTIONS
    return CANONICAL_DESCRIPTIONS

规则：

条目以 get_schemas 返回的端点/schema 名称为键（与 ENDPOINTS 匹配），而不是带前缀的仓库表名。
源描述来自官方 API 文档，而不是猜测。部分覆盖是可以的 —— 任何缺失的端点、列或表级 description 都会回退到大模型，大模型会获得源名称、端点、docs_url 和列数据类型。
可选，且仅对固定 schema 的源有意义。SQL 源（任意用户 schema）不提供任何内容 —— 基础 hook 返回 {}。
不要改动 source.py/settings.py 的传输逻辑 —— 这纯粹是新增的元数据。

将表目录发布到公开文档

posthog.com 文档通过 <SourceTables /> 组件渲染 Supported tables 部分，该组件的数据来自 public_source_configs API，而该 API 会对每个 source 调用 get_documented_tables()。基础实现仅在 source 主动启用时，才会列出 get_schemas（与 canonical_descriptions 合并）中的表：

class MySource(SimpleSource[MySourceConfig]):
    lists_tables_without_credentials = True  # static endpoint catalog — safe for public docs

只有当 get_schemas 遍历静态端点目录且无 I/O——无网络、无 DB、无凭证（常见的固定 schema SaaS 模式： for endpoint in ENDPOINTS）时，才将其设为 True。该端点会构建一个占位配置，并在没有真实凭证的情况下调用 get_schemas，因此凡是需要连接以发现 schema 的 source（SQL、文件存储、MongoDB、会列出账户的广告平台）都必须保持此值为 False（默认值）——否则它会尝试连接空主机、挂起或关闭 DB 会话。当为 False 时，文档会改为渲染一条通用的“discovered from your account”说明。

表列表越丰富，文档越好——因此请将其与 canonical_descriptions.py （表/列描述）搭配使用。通过 API 验证渲染输出： GET /api/public_source_configs → 你的 source → tables。

Source 类别与关键词

每个 source 必须在其 SourceConfig 上设置 category——它会在新建 source 向导目录（类别栏 + 磁贴网格）中对 source 进行分组。如果任何已注册的 source 没有类别，测试（tests/test_source_categories.py）就会失败，因此这是必选项。从 products.warehouse_sources.backend.facade.source_config 导入 SourceConfig 和 DataWarehouseSourceCategory，并从 facade 的 ExternalDataSourceType 导入 name：

from products.warehouse_sources.backend.facade.source_config import DataWarehouseSourceCategory, SourceConfig
from products.warehouse_sources.backend.types import ExternalDataSourceType
...
return SourceConfig(
    name=ExternalDataSourceType.STRIPE,
    category=DataWarehouseSourceCategory.PAYMENTS___BILLING,
    keywords=["billing", "subscriptions"],
    ...
)

选择最接近的单个分类。枚举成员如下（注意标签含 " & " 时使用三下划线）：

DATABASES — OLTP/OLAP 数据库、数仓、数据流（Postgres、Snowflake、BigQuery、Kafka 等）
FILE_STORAGE — 对象/文件存储与文件传输（S3、Azure Blob、GCS、Google Drive、SFTP 等）
ADVERTISING — 广告平台与移动归因（Google Ads、Meta Ads、Reddit Ads、Adjust 等）
MARKETING___EMAIL — 电子邮件/短信/营销自动化（Klaviyo、Mailchimp、Braze、SendGrid 等）
CRM — CRM 与销售情报（HubSpot、Salesforce、Attio、Pipedrive、ZoomInfo 等）
SALES — 销售互动/赋能、合同（Salesloft、Outreach、Gong、DocuSign 等）
CUSTOMER_SUPPORT — 帮助台/支持/CX（Zendesk、Intercom、Freshdesk、Front 等）
PAYMENTS___BILLING — 支付处理与订阅计费（Stripe、Chargebee、PayPal 等）
FINANCE___ACCOUNTING — 会计/ERP/费用/支出（QuickBooks、Xero、NetSuite、SAP ERP 等）
ANALYTICS — 产品/Web/营销分析与实验（Amplitude、Mixpanel、GA 等）
ENGINEERING___MONITORING — 开发工具、CI、错误/正常运行时间监控、功能开关、身份/认证（GitHub、Datadog、Sentry、LaunchDarkly、Auth0 等）
PRODUCTIVITY — 项目管理、文档、表单、日程安排（Notion、Airtable、Jira、Linear、Typeform 等）
HR___RECRUITING — HRIS/ATS/薪资/人员（Ashby、Greenhouse、BambooHR、Workday、Gusto 等）
COMMUNICATION — 消息/会议/电话/社交（Slack、Zoom、Microsoft Teams、Twilio 等）
E_COMMERCE — 在线商店/商务（Shopify、WooCommerce、BigCommerce 等）

类别列表由后端拥有：products/warehouse_sources/backend/facade/source_config.py 中的 DataWarehouseSourceCategory 是唯一事实来源，hogli build:openapi 会根据它重新生成前端类型。添加新类别意味着要修改该枚举——不要凭空造出临时字符串。

keywords 是一个可选的小写搜索别名列表——仅当 source 具有用户可能输入的常见缩写或其他拼写时才添加（例如 ["ga4", "ga"]、["sql server"]、["facebook ads"]）。当名称本身已明显匹配时请跳过；不要添加噪声。

代码库检测

已发布的源（没有 unreleasedSource=True）必须设置 codebase_detection 类属性。PostHog 向导会用它判断客户的代码库是否使用了该厂商，从而推荐该源并发现 PostHog 的替代方案。使用 products/warehouse_sources/backend/facade/codebase_detection.py 中的 detect(...)。列出该厂商在各生态中的官方 SDK 包名、该厂商文档中提及的 .env 键名（使用锚定正则表达式），以及其脚本标签加载自的 CDN 主机。除非源的字段需要 OAuth、文件上传或账户选择器，否则向导会从终端创建源。如果用户仍应前往应用中操作（例如因为凭据不易查找），请将类属性 setup_requires_ui = True。如果没有任何包、键或脚本能标识该厂商，请使用 NOT_DETECTABLE。对于已发布但未设置该属性的源，以及设置了该属性的未发布源，tests/test_codebase_detection.py 都会报错。构建镜像时会将这些声明渲染到向导下载的目录中，因此无需更新生成的文件。

自驱动收件箱候选资格（issues / tickets / conversations）

某些源也是 自驱动收件箱 的候选源——该功能会监控已同步的 可采取行动的记录 表，并将发现结果发送到 PostHog 桌面收件箱。目前已支持：GitHub、 Linear、Zendesk、pganalyze 和 Jira。

判断信号的是你同步的表，而非厂商：如果某个源的表是由人工（或 Agent）逐条分类处理的记录流，例如 issues、tickets 或 conversations 表，那么该源就是收件箱候选源。这些表属于支持/帮助台（CUSTOMER_SUPPORT）、问题跟踪和监控（ENGINEERING___MONITORING），以及部分项目工具（PRODUCTIVITY）类别。分析、计费、广告平台、CRM 和原始数据库源都不是收件箱候选源——它们同步的是供查询的数据，而非需要采取行动的工作队列。如果你正在构建的源没有这类表，则无需处理此项。

将源接入收件箱是独立且增量的一项工作，有其专属技能—— /adding-inbox-sources——不会改变本技能的交付内容。只有在数据仓库源已存在后才能进行这项工作（这正是本技能的产出），因此请先构建并发布该源。这项技能会涉及三个部分：本仓库中的服务端“signals scout”发射器、注册表条目和 SignalSourceProduct 枚举（位于 products/signals/backend/）；单独的 posthog/code 仓库中的收件箱 UI；以及 PostHog/context-mill 中的 npx @posthog/wizard self-driving 入门流程。开始之前请阅读 /adding-inbox-sources——这些接入工作都不属于源本身的 products/warehouse_sources/ 代码。

供应商 API 版本元数据

每个 source 都会（在 source 类主体上，与 lists_tables_without_credentials 并列）声明三个类属性，用于描述供应商的 API 版本。框架（common/base.py）会记录每个 ExternalDataSource 所运行的版本，以便旧的版本锁定仍能正常工作，并能暴露弃用信息； sources/tests/test_source_versions.py 会在每个已注册的 source 上强制校验以下不变量，因此新 source 如果这些写错，CI 就会失败。

两种情况：

供应商提供了真实、可锁定的 API 版本 —— 例如 URL 路径段（/v3/、/2/）、必填的版本头值（带日期的 2022-11-28）、带日期的查询/版本参数，或具名发布版本。声明全部三项：
```
class MySource(SimpleSource[MySourceConfig]):
    supported_versions = ("v3",)          # opaque vendor labels — never parsed or ordered
    default_version = "v3"                 # stamped onto newly created sources; must be in supported_versions
    api_docs_url = "https://vendor.example/docs/api"   # API reference or changelog page (https, not the marketing site)
```
基于供应商最新的正式可用版本构建 source，并锁定该版本。 新 source 从一个版本起步，每个连接它的客户都会落到该版本，因此基于旧版本发布就意味着发布了一个日后必须有人执行的迁移。两条规则，且必须一致：
1. 针对供应商提供的最新 GA 版本编写请求代码。
2. 声明代码实际调用的版本（基础 URL 路径、版本头，或 settings.py / {source}.py 中的版本常量）。绝不声明代码未发送的版本 —— 这个锁定是框架无法检测的谎言，并且会让每个客户的弃用警告和升级路径都出错。
如果你无法使用最新版本 —— 它处于 preview/beta/unstable/RC 阶段、需要通过申请或付费层级才能使用，或者其响应结构尚未实现 —— 就基于你实际能调用的最新 GA 版本构建，锁定该版本，并在类的注释中说明原因。“最新”指最新稳定版：不要锁定 Shopify 的 unstable、供应商的 -rc 渠道，或文档仍标注为 preview 的版本。

代码库中已有的示例：Anthropic ("2023-06-01",)（带日期的 anthropic-version 头）、 ActiveCampaign ("v3",)（/api/3 路径段）、Alguna ("2026-04-01",)（带日期的版本头）。之后获得第二个版本的 source 按从旧到新声明 —— GitHub ("2022-11-28", "2026-03-10")、 HubSpot ("v3", "2026-03") —— 但那是 /warehouse-source-new-version 技能的职责，不是本技能的。
供应商没有有意义的 API 版本管理 —— 只设置 api_docs_url；将 supported_versions / default_version 保留为框架默认值（("v1",)，即 UNVERSIONED_API_VERSION 哨兵值）。一个从未变过且并非文档化版本选择的裸 /v1/ 就属于这种情况。

规则：

default_version 必须等于 supported_versions 中的唯一一项，且 api_docs_url 必须为 https://。
使用供应商的确切版本字符串；绝不臆造。
绝不要基于供应商已弃用或已给出终止日期的版本发布新 source。 一个全新的 source 如果其 deprecated_versions 条目覆盖了它唯一的版本，那就是 bug —— 这意味着该 source 是针对错误的版本编写的。如果 default_version 已被弃用，test_source_versions.py 会让构建失败。
优先选择指向供应商版本管理/变更日志页面的 api_docs_url，而非通用的 API 落地页 —— 那是下一个版本发布公告的地方，也是下一个人重新锁定版本前会查看的地方。
不要在传输/请求层硬编码回退版本 —— 应从 source 类解析它（self.resolve_api_version(inputs.api_version)），它已经会回退到 default_version。
之后为新供应商版本添加支持，或弃用旧版本，属于 /warehouse-source-new-version 技能 —— 不是本技能。

源字段（用户填写的表单）

在 get_source_config.fields 中定义。所有字段类型都位于 products/warehouse_sources/backend/facade/source_config.py，并在 products/warehouse_sources/backend/temporal/data_imports/sources/common/base.py 中通过 FieldType 进行联合。

SourceFieldInputConfig — 基础输入（text、email、number、password、textarea）。渲染为 <LemonInput />。
SourceFieldSwitchGroupConfig — 开关，可展开一组子字段。用于可选功能块。
SourceFieldSelectConfig — 下拉框。选项可以携带选中时显示的子 fields（用于替代认证方式 — 例如 API key 与 OAuth）。
SourceFieldOauthConfig — 通过 Integration 模型实现的 OAuth。参见 OAuth 章节。
SourceFieldFileUploadConfig — 文件上传（JSON）。使用 keys=["..."] 允许列表或 "*"。
SourceFieldSSHTunnelConfig — 渲染 SSH 隧道子字段；向配置添加带辅助工具的 ssh_tunnel: SSHTunnel。

指南：

多种认证方式 → 使用 SourceFieldSelectConfig，并为每个选项配置子 fields。
可选开关 → 使用 SourceFieldSwitchGroupConfig。
机密字段必须使用 SourceFieldInputConfigType.PASSWORD。序列化器会根据字段定义自动推导敏感与非敏感键 — 你无需在其他地方维护允许列表。

实现 `source_for_pipeline`

直接返回 SourceResponse。不要对新源使用 dlt_source_to_source_response — DLT 正在被移除。

优先按 API 返回的数据形状产出数据。不要自定义 dataclass，不要进行繁重解析。产出 dict、list[dict]（在可能时优先）或 pyarrow.Table。管道会为你缓冲并分批。

默认产出原始 dict / list[dict]，让管道为你分批。 管道已在 5000 行 / 200 MiB 阈值处运行 Batcher（pipelines/pipeline_v3/pipeline.py），因此常见情况无需自带 batcher。只有在你已经拥有 arrow 形状的数据时（例如 ClickHouse 适配器）才使用 pyarrow.Table。源_可以_以更小的阈值实例化自己的 Batcher（例如 chunk_size=2000, chunk_size_bytes=100 * 1024 * 1024，klaviyo 和其他约 70 个源就是这样做的），当它有意为大/宽行追求更紧凑的内存占用时 — 这是一种有效选择，而非默认做法。要避免的是第二个_全尺寸_ batcher，那只会带来无收益的双重缓冲。

对于 pyarrow 表，将内存中的行数限制在约 200 MiB 或约 5000 行。使用来自 products/warehouse_sources/backend/temporal/data_imports/pipelines/core/arrow_utils.py 的 table_from_iterator() / table_from_py_list() 等辅助工具。

URL 构造： 查询字符串使用 urllib.parse.urlencode。不要使用 requests.Request(...).prepare().url — PreparedRequest.url 的类型是 Optional[str]，而典型的变通方案（prepared.url or f"..."）带有一个不可达的回退。urlencode 更短、无依赖，并且对于 ASCII 安全参数产生相同的输出。

可恢复的 source 模式

@dataclasses.dataclass
class MyResumeConfig:
    next_url: str  # or cursor, offset, time window — whatever the API uses

class MySource(ResumableSource[MySourceConfig, MyResumeConfig]):
    def get_resumable_source_manager(self, inputs: SourceInputs) -> ResumableSourceManager[MyResumeConfig]:
        return ResumableSourceManager[MyResumeConfig](inputs, MyResumeConfig)

    def source_for_pipeline(
        self,
        config: MySourceConfig,
        resumable_source_manager: ResumableSourceManager[MyResumeConfig],
        inputs: SourceInputs,
    ) -> SourceResponse:
        return my_source(..., resumable_source_manager=resumable_source_manager)

在 transport 函数中：

resume = manager.load_state() if manager.can_resume() else None
url = resume.next_url if resume else initial_url

while True:
    data = fetch_page(url)
    next_url = data.get("links", {}).get("next")
    if next_url:
        manager.save_state(MyResumeConfig(next_url=next_url))  # stage before the yield it covers
    # yield batch
    if not next_url:
        break
    url = next_url  # advance before the next fetch, otherwise we loop on the same page

在 yield 其所覆盖的批次之前保存状态。save_state 只暂存游标；流水线会在该批次写入后将其提交到 Redis，因此崩溃后会精确地从最后写入的批次之后恢复。不要在 yield 之后保存。在 worker 关闭时，流水线会在控制权返回 source 之前结束本次尝试，因此 yield 之后保存的状态会在最后一个批次中丢失：下一次尝试会再次读取该批次（merge 会按主键去重，而恢复的完整刷新会将其追加两次），并且写入一个批次或更少的尝试不会保留任何进度。尚未 yield 任何内容的 source（例如在轮询导出任务 id 之前先将其持久化的 source）会在 with manager.committing(): 内部暂存，该块结束时会提交。

已保存的游标只有在流水线确认之后才能提交。流水线会在 source 将下一个条目交给它、结束或到达安全点时进行确认。因此，在最后一个 yield 之后保存的游标在 source 抛出异常时不会持久化，因为 source 仍可能持有游标所跳过的行。下一次尝试会从最后一个 yield 的游标继续。因分页或时间预算而结束自身尝试的 source，会在抛出异常前直接调用 manager.safe_point()，且其本地缓冲区为空，以保留其最后一个游标。

在 source 可能发出许多不返回任何行的请求的地方调用 manager.safe_point()：空的 delta 页、没有子项的 fan-out 父项、没有评论的页。流水线仅在条目到达时检查 worker 关闭，因此一连串空响应否则会让 worker 在整个优雅关闭超时期间被占用，并且其游标永远不会提交。在安全点，流水线可以将本次运行交给另一个 worker，并在没有等待写入的内容时提交已暂存的游标。仅在从已暂存游标恢复不会丢失任何行的地方调用它：游标覆盖的每一行都已 yield，并且没有任何行位于本地缓冲区中。参考：convex/convex.py 中的 document_deltas、stripe/stripe.py 中的 sparse-sweep checkpoint、notion/notion.py 中的 _page_fan_out。 rest_source 框架会自行在每页之后以及每次重试等待之前到达安全点，但仅当 SourceResponse.items 直接返回框架的 Resource 时才会如此。包装它的 source 不会获得框架的安全点，因为包装器可能会缓冲行。同样的条件决定 resume_hook 何时运行。当 items 直接返回 Resource 时，hook 会在页面到达流水线之前运行，因此页面及其游标会一起提交，交接时不会重复任何行。当 source 包装 Resource 时，hook 会在包装器请求下一页时运行，因此交接会再次读取最后一页。尽可能直接返回 Resource：使用 data_map、add_map 和 add_filter 进行行变更。一个将每页原样传递且不持有任何行的包装器，可以通过返回 Resource(wrapper, name=..., hints=resource._hints) 来保留框架行为（参见 anthropic/anthropic.py 中的 usage report）。不要在 resume_hook 中调用 safe_point() 或 commit()。

源契约

一个关闭的 worker 会把每个正在运行的导入交给另一个 worker。每个带有提取代码的源都必须让这一点成为可能：

有界调用：每个请求都有读取超时，一个请求连同其重试和等待最多占用 worker 5 分钟，在收到带有较大 Retry-After 的 429 之后也是如此。
每个请求都有一个安全点：源连续发出的请求不超过 3 个，且其间没有让出，也没有安全点。
恢复状态在它所覆盖的 yield 之前暂存。

sources/tests/test_source_contract.py 会让注册表中的每个源针对一个会停滞、限流并返回空页面的假 HTTP 服务器运行。它将结果与 sources/tests/source_contract_baseline.txt 进行比较，后者列出了每个未通过某项条件或假服务器无法驱动的源。新源必须通过；基线只能变短。 posthog/test/repo_invariants/test_warehouse_source_static_contract.py 和 test_resume_state_staged_before_yield.py 会在每个源的源文本中检查同一契约：没有裸 time.sleep，没有不带 timeout= 的会话调用，没有在框架 Resource 外面不带安全点的包装器，以及 yield 之后没有 save_state。

pytest products/warehouse_sources/backend/temporal/data_imports/sources/tests/test_source_contract.py posthog/test/repo_invariants/test_warehouse_source_static_contract.py posthog/test/repo_invariants/test_resume_state_staged_before_yield.py

当某个修复使某个条目不再正确时，重新生成基线并提交：

SOURCE_CONTRACT_WRITE_BASELINE=1 pytest products/warehouse_sources/backend/temporal/data_imports/sources/tests/test_source_contract.py -p no:xdist
python posthog/test/repo_invariants/test_warehouse_source_static_contract.py
python posthog/test/repo_invariants/test_resume_state_staged_before_yield.py

Webhook 源模式

实现 webhook_template，返回一个 HogFunctionTemplateDC，用于转换传入的 webhook 负载。
实现 webhook_resource_map，将我们的 schema 名称映射到外部对象类型。
如果 API 允许通过程序管理 webhook，则实现 create_webhook、delete_webhook、get_external_webhook_info。否则返回失败结果，并提供 webhookSetupCaption 说明手动设置方式。
向 SourceConfig 添加 webhookFields，用于设置后的输入（例如签名密钥）。
在 source_for_pipeline 中，调用 self.get_webhook_source_manager(inputs)，并将其迭代器与拉取迭代器一起传入，这样一次同步就能拉取历史行 + webhook 投递的行。
仅在 webhook 实际可行的端点上（通常是增量/仅追加的端点）填充 SourceSchema.supports_webhooks=True。
使用 table_transformer 在 webhook 批次内去重。 WebhookSourceManager.get_items() 接受一个可选的 table_transformer: Callable[[pa.Table], pa.Table]，在原始 webhook 负载被反序列化为行字典之后应用。Delta 合并只会在跨同步时（基于 primary_keys）去重，而不会在单个源批次内去重——因此当一个批次可能携带同一对象的多个事件时（例如先 customer.created 后 customer.updated），传入一个只保留每个 id 最新版本的转换器。参考：stripe/stripe.py 中的 _webhook_table_transformer，通过 stripe_source 中的 webhook_source_manager.get_items(table_transformer=_webhook_table_transformer) 接入。它按 object.id 对行分组，保留事件 created 时间戳最大的那一行，并重建出与底层对象形状一致的表（可直接按 primary_keys=["id"] 合并）。

多 schema SQL 数据库源

SQL DB 源（目前包括 Postgres、MSSQL、Snowflake、Redshift）可以在一个连接中导入来自每个命名空间（schema）的表：命名空间字段留空会发现所有非系统命名空间中的表，向导会按命名空间对它们分组，同步会为每个 namespace.table 写入一个仓库表。参考实现：postgres/postgres.py + PostgresImplementation；共享接缝位于 common/sql/。

能力标记是源的 schema 字段在 get_source_config 中为可选（required=False）——is_multi_schema_capable_sql_source()（products/data_warehouse/backend/sql_warehouse_migration.py）以此为依据，因此将该字段改为可选就会开启 viewset 迁移行为。将 None / "" / 空白视为“所有命名空间”（common/sql/location.py 中的 normalize_namespace），并且永远不要生成 WHERE table_schema = ''。

让 SQL 源达到多 schema 对等性的检查清单：

命名空间字段可选 — 在 schema 字段上设置 required=False，重新运行 pnpm run generate:source-configs。保持 database 为必填：database/catalog 在每个连接中保持固定。
多命名空间发现 — 在 get_columns、get_primary_keys、索引/行数/外键辅助函数中：当命名空间为空时，去掉 WHERE table_schema = <ns> 谓词（排除系统命名空间，如 information_schema、pg_catalog、sys），并返回限定显示名（namespace.table）。当该字段已设置时，保留单命名空间快速路径。
实现 get_source_metadata — 返回以限定显示名为键的 SourceMetadata(catalog_by_table, schema_by_table, table_name_by_table)。SQLSource.get_schemas 会将其标记到每个 SourceSchema 上，reconcile_schema_metadata 会将其持久化到 ExternalDataSchema.sync_type_config["schema_metadata"]。
在 build_pipeline 中按行路由 — 使用 resolve_source_location（common/sql/location.py）解析 (schema, table_name, response_name)：按行元数据 → 点号名称自愈 → 配置命名空间。针对解析出的 schema + 未限定的表运行 SQL；设置 SourceResponse.name = response_name（dwh_storage_key or schema.name，规范化后）——绝不要使用裸表名，否则该行的 Delta 路径会改变并使其已同步数据成为孤儿。
将解析出的命名空间贯穿每个流式/统计辅助函数 — 表元数据、行统计、平均行大小、分区设置、块大小、主键查找都接收 (schema, table)。漏掉一个就会静默降级（无分区 / 统计错误）。
永远不要把带点号的显示名传给标识符引用器 — quote("a.b") 会生成一个错误的标识符。先拆分为 (schema, table)，再使用 quote_qualified（common/sql/identifiers.py）。
命名层是派生的，绝不存储 — 显示名 analytics.users；S3/Delta 子目录 analytics_users（规范化后的 response_name）；HogQL 表 {prefix}_analytics_users。唯一存储的例外是 dwh_storage_key，它将迁移后的旧行固定到其原始 Delta 路径。
旧版迁移由能力驱动，而不是由源类型门控 — 当用户在现有单 schema 源上清空命名空间时，sql_warehouse_migration.py 会将行重命名为限定形式并标记 dwh_storage_key，在无需重新同步的情况下保留已同步数据。不要向共享层添加 source_type == "..." 分支。
测试 — 两个具有相同表名的命名空间端到端保持不同；空命名空间发现会排除系统命名空间；按行路由会命中正确的命名空间；旧版单命名空间源仍可工作；迁移后的行保留其旧版 Delta 路径。

发现成本：validate_credentials 和 database_schema 会在没有名称过滤器的情况下运行发现，因此在拥有数百个 schema 的 catalog 上留空命名空间时，绝不能按命名空间对每张表发起查询——请批处理列表查询或限制枚举（参见 Snowflake 的 SHOW PRIMARY KEYS 处理）。

出站 HTTP 必须经由受跟踪的传输层

来自 products/warehouse_sources/backend/temporal/data_imports/sources/** 的每一次 HTTP 调用都必须经由 make_tracked_session()（来自 products.warehouse_sources.backend.temporal.data_imports.sources.common.http）。受跟踪的会话会将 team_id、source_type、 external_data_source_id、external_data_schema_id 和 external_data_job_id 附加到每一次出站请求的日志行和 OTel 指标上，并参与可选择加入的样本采集。

对于直接使用 requests 的情况：make_tracked_session(headers=..., retry=...) 返回一个 requests.Session。请使用 session.get/post/...，而不是模块级的 requests.get/... 快捷方法。
从采集的样本中脱敏密钥。 向 make_tracked_session(...) 传入 redact_values=(api_key, token, ...)，以便受跟踪的传输层在记录的 URL、请求头和采集的请求体中遮蔽这些字面值——这对于那些出现在查询参数或奇怪请求头名称中的密钥尤为重要。rest_source 的认证类会自动执行此操作（每个都实现了 secret_values()）；只有使用原始会话的 source 才需要自己传入 redact_values。
对于已经经由 rest_source.RESTClient 的 source：它默认自动使用受跟踪的会话，无需更改。
对于接受 session/HTTP-client 钩子的厂商 SDK（Stripe 的 RequestsClient(session=...)、 gspread 的 authorize(credentials, session=...)、BigQuery 通过 AuthorizedSession + TrackedHTTPAdapter），请注入一个。参考模式见 stripe/stripe.py、google_sheets/google_sheets.py 和 bigquery/bigquery.py。
对于没有注入接缝的厂商 SDK（目前有：bingads、linkedin-api 的 RestliClient），请添加一个 # nosemgrep: data-imports-http-transport-... 编译指示，并附上一行原因，同时将该 source 记录为 ⚠️ Vendor SDK（在 SOURCES.md 中）。
gRPC SDK 不豁免——它们有自己的受跟踪传输层（见下文）。

CI 通过 .semgrep/rules/security/data-imports-http-transport.yaml 强制执行此要求。该规则禁止在 sources/** 内直接使用 requests.Session()、requests.<verb>(...) 和 httpx.Client/AsyncClient/<verb>。仅用于类型标注的导入（from requests import Response、from requests.exceptions import HTTPError）仍然允许。

出站 gRPC 必须经由受跟踪的 gRPC 传输层

来自 sources/** 的 gRPC 调用通过 products.warehouse_sources.backend.temporal.data_imports.sources.common.grpc 的客户端拦截器进行，这些拦截器会将相同的 JobContext 标签附加到日志和 OTel 指标（data_import_grpc_*）上，并为可选择加入的样本采集（protobuf → 已脱敏的 JSON）提供数据。有两个接缝：

对于接受 interceptors= 列表的 SDK（google-ads 的 GoogleAdsClient.get_service(...)），请在每一次 get_service 调用时传入 interceptors=tracked_interceptors(host)——google-ads 会为每次调用重建 channel，因此每次都必须重新提供拦截器。参考：google_ads/google_ads.py。
对于接受 channel= / transport= 的 SDK（BigQuery Storage Read API），请构建携带凭据的 channel，用 make_tracked_channel(channel, host=...) 包装它，然后交给该 transport。参考： bigquery/bigquery.py:bigquery_storage_read_client。

CI 通过 .semgrep/rules/security/data-imports-grpc-transport.yaml 强制执行此要求，该规则禁止在 sources/** 内（common/grpc/ 包和两个参考 source 文件之外）直接使用原始的 grpc.*_channel(...) 以及直接构造 BigQueryReadClient(...) / GoogleAdsClient(...)。运维人员通过 python manage.py warehouse_sources_capture_grpc_samples enable ... 来开启样本采集。

更新 SOURCES.md

products/warehouse_sources/backend/temporal/data_imports/sources/SOURCES.md 是所有已注册数据源的清单，记录了它们的通信方式，以及其出站流量是否被追踪。每当你进行以下操作时，请在同一 PR 中更新它：

添加新数据源 —— 最初作为 Scaffolded 条目；一旦你交付了可用的同步逻辑，就把它移入 Implemented 表。
实现之前已搭建脚手架的数据源 —— 将该行移入 Implemented 表，并填写通信方式、主要库和传输追踪状态。
迁移厂商 SDK 以注入被追踪的会话 —— 将该数据源从 ⚠️ Vendor SDK 改为 ✅。
切换数据源的协议 —— 例如将 REST 换成 gRPC、在拉取 API 之外增加 webhook 支持，或从 requests 迁移到厂商 SDK。同时更新通信方式和传输追踪两列。

在每个表内保持条目按字母顺序排列。scaffolded 列表每行一个数据源（每项一个 bullet，同样按字母顺序），这样添加或移除数据源只会影响它自己那一行，避免与并发的 PR 产生冲突——不要把它重新合并成逗号分隔的段落。如果你添加了一个部分追踪的数据源，还要追加一条简短的 "Notes on partially-tracked sources" 条目，说明是什么阻碍了追踪（例如厂商 SDK 没有 session/interceptor 接缝）。

基类能力标志与 API 版本管理

common/base.py 暴露了一些类级标志，大多数 API 数据源保持其默认值，但在适用时很重要：

supports_column_selection（默认 False；SQLSource 设为 True）—— 数据源是否通过 SELECT 投影支持 enabled_columns。
supports_row_filters（默认 False）—— 数据源必须为 True 才能应用已保存的 row_filters；否则过滤器会被忽略。
has_managed_hogql_schema（默认 False；对 Stripe、Paddle、Zendesk 等营收分析类数据源为 True）—— 固定的字段集支撑托管的 HogQL schema，这会禁用列选择。新的计费/营收数据源如果忽略这一点，可能会破坏营收分析。
cleanup_cdc_resources_on_deletion —— 用于创建 CDC 资源的数据源的最佳努力清理钩子；仅当你的数据源创建此类资源时才覆盖。

对于厂商 API 版本管理（supported_versions、default_version、api_docs_url、 deprecated_versions、resolve_api_version()），请使用专门的 warehouse-source-new-version skill——不要手写版本处理逻辑。

必需的编码约定

使用 @SourceRegistry.register 注册。
继承 SimpleSource[GeneratedConfig]，除非需要可恢复/webhook 行为。
API 数据源通常应在 endpoint resources 中返回 table_format="delta"。
primary_keys 是 endpoint 特定的（在 settings.py 中声明，并不总是 id）。当没有单个字段唯一时，使用复合键。键必须在整张表中唯一，而不是在每个父级中唯一：扇出子 endpoint 会聚合来自每个父级的行，因此除非 API 明确记载全局唯一，否则要在键中包含父标识符（例如 ["form_id", "token"]）。非唯一的键会在 Delta 表中种下重复行，之后每次 merge 都会多重匹配它们——merge 会每次同步都变慢，直到 pod OOM。
尽可能为新数据源添加分区：
- API 数据源：partition_mode="datetime" 配合一个稳定的 datetime 字段。
- 数据库数据源：partition_count 和 partition_size。
选择一个不会变化的分区键——created_at、dateCreated、firstSeen。绝不要使用 updated_at 或 lastSeen。
为已知的永久性失败（401/403、无效/过期的凭据、缺少 scope）添加 get_non_retryable_errors()。
保持注释最少，仅在意图不明显时添加。
Python 导入放在模块顶部，而不是函数内部（除非需要打破循环导入）。

增量同步指南

仅当 API 暴露服务端时间戳过滤条件（<field>_gte、since、modified_after 等）时，才设置 supports_incremental=True。 在 Python 中拉取每一页并跳过已见行的“客户端游标”不是增量的——每次运行仍然会访问每一页，因此“增量”同步的 API 开销最终与全量刷新完全相同。如果 API 没有服务端过滤条件，就只提供全量刷新。
如果 API 支持服务端时间过滤，就使用它，并从 db_incremental_field_last_value 进行映射。
遵循 inputs.incremental_field —— 这是用户在 schema 设置中选择的游标字段。每个 endpoint 的 INCREMENTAL_FIELDS 是_可公布的选项_菜单；不要伸手去取 INCREMENTAL_FIELDS[endpoint][0] 来挑选默认值并静默覆盖用户的选择。
每个 endpoint 的排序枚举各不相同。 不要全局硬编码 ?sorting=created_at（或类似值）。对照 API 规范并通过针对实时 API 的 curl 冒烟测试，逐一核实每个列表 endpoint 允许的排序值——API 经常记录一套选项却静默拒绝另一套，或对某些资源使用不同的时间戳列。
分页时，在稳定的单调字段上显式传入 ?sorting=。 对于增量源，请求排序必须匹配 SourceResponse.sort_mode（通常是 "asc"；仅当 API 强制要求时才用 "desc" —— 参见 stripe/stripe.py、github/settings.py），以便流水线的游标水位线正确推进。对于全量刷新源，显式排序可防止在 API 的隐式默认排序不稳定、或随同步过程中插入行而变动时，出现分页边界的跳过/重复。
如果 API 仅支持游标分页，只要可靠，仍然声明增量字段，并让合并语义去重。
sort_mode 必须与行实际到达的顺序一致——要验证，不要假设。 流水线信任 sort_mode="asc" 才能在每批之后对增量水位线做检查点，并允许同步中途安全关闭 worker；在 API 返回最新优先时却声明 asc 会破坏水位线并破坏恢复语义。检查 API 的_默认_排序（当你无法传入 sort 时它会生效），并记住游标分页常常拒绝或完全忽略排序参数。
仅当 endpoint 确实无法返回升序时，才使用 sort_mode="desc"。对于降序源，处理 db_incremental_field_earliest_value，以便在新行之前先回溯更早的行（参见 Stripe）。
增量分页必须在水位线处终止。 有些 API 拒绝将时间窗口过滤与游标分页混用，因此只有第一页受窗口限制，后续页会无界地回溯历史。如果服务器无法在每一页上保持该过滤条件，分页器必须在客户端停止，一旦整页都早于 db_incremental_field_last_value（参见 typeform/typeform.py:TypeformResponsesPaginator）—— 否则每次增量同步都会重新拉取并重新合并每个父级的完整历史，这既是 API 开销 bug，也是每次同步的内存放大器。
未知 endpoint 先默认全量刷新；只有在确认稳定的过滤字段和 API 排序语义后，才启用增量。
依据响应 schema 而非 endpoint 名称来确认分区键。

API 行为验证清单

在最终确定端点逻辑之前，请从文档以及使用 curl 针对实时 API 进行验证（不要只看文档——API 经常静默忽略未知参数或文档中过时的枚举值）：

响应结构：列表 vs 对象 vs 包装数据（{"data": [...]}）。
分页：Link 头 vs 响应体游标 vs offset/page；如何标识下一页终止。
排序保证：时间字段的升序/降序/未定义，以及如果不传排序参数时 API 的_默认_排序。如果使用游标（before/after token）分页，请确认 API 是否允许同时传入 sort 和时间窗口参数——很多 API 会拒绝或忽略它们，这决定了你的 sort_mode 以及增量同步时分页如何终止。
主键唯一性范围： id 是全局唯一，还是仅在其父资源内唯一？对于扇出子资源，除非文档另有说明，否则假设为按父资源唯一，并将父 id 放入复合键中。
每个端点的排序枚举： 每个列表端点接受哪些 sorting= 值？有些 API 会根据资源不同而改变允许的枚举值。使用 curl 确认你打算传入的值返回 200，并用未来日期截止值探测，以确认时间戳过滤器是被遵守还是被静默忽略。
服务端时间戳过滤： <field>_gte / since / modified_after 是否真正执行过滤，还是 API 接受但忽略它？通过传入未来日期并检查该行是否被排除来测试。
速率限制头（窗口重置时间戳、并发限制）。
字段稳定性：候选增量/分区字段是否会随时间变化。

如果未记录在文档中，请保持解析/合并逻辑保守，并添加简短的代码注释说明不确定性。

端点清单工作流

在扩展覆盖范围之前构建端点清单（路径、认证作用域、粒度、分页方式、主键形态、增量候选字段）。
将其保存在源码本地文档中（例如 products/warehouse_sources/backend/temporal/data_imports/sources/<source>/api_inventory.md）。
分阶段添加端点：组织级列表端点 → 项目级扇出 → 带有限分页的子/扇出端点。

顶层端点（组织/账户级别）

在 settings.py 中声明端点元数据（path、primary_key、incremental_fields、partition_key、sort_mode）。
通过单一资源配置辅助函数构建；保持传输分支最少。
端点参数保持声明式（limit、必需过滤器）。
仅在增量语义可靠时合并写入处置；否则全量替换。

分页提示

有些 API 使用 Link 头中的游标分页——检查 rel="next" 和任何结果标志。
当跟随响应头中的完整游标 URL 时，在分页器的 update_request 中清除请求参数，以避免重复的查询参数。
对于父/子扇出，为每个父资源设置硬性页数上限，以避免无界扫描。
达到页数上限时输出结构化日志（包含资源名称和父标识符）。

重试与限流策略

make_tracked_session() 和 rest_source.RESTClient 已在传输层重试 429 和瞬时 5xx，并遵循 Retry-After。不要在获取逻辑外再包一层 tenacity @retry 来处理状态码——它会叠加（例如 3 次传输层尝试 × 5 次 tenacity 尝试），这是现有数据源中最常被复制的错误。如果使用框架或 tracked session，状态码重试已经处理好了；不要自己写。
由一层负责某个请求的重试。 RESTClient 会重试自己的请求（默认 5 次尝试），并对其关闭适配器重试。在 tracked session 上的自定义调用保留适配器重试（3 次重试，仅限 GET/HEAD/OPTIONS）。
重试等待是有上限的。 单个请求最多花 DATA_WAREHOUSE_SOURCE_RETRY_BUDGET_SECONDS（600 秒）在重试上。当服务端要求的等待时间超过 DATA_WAREHOUSE_SOURCE_MAX_RETRY_AFTER_SECONDS（300 秒）时，RESTClient 会以可重试错误让请求失败。对于限流窗口更长的供应商，在客户端配置中设置 retry_after_max_seconds 和 retry_budget_seconds。
不要用 time.sleep 做退避或限流等待。 改为调用 common/interruptible_wait.py 中的 interruptible_wait()。当 worker 开始关闭时它会结束等待，运行可以恢复。仅当暂存游标未覆盖数据源仍持有的任何行时，才传递 safe_point=。
仅当传输层不覆盖某个条件时才添加 tenacity——例如应用层返回的、不是可重试 HTTP 状态的“仍在处理”响应体。如果这样做，请关闭传输层重试以避免叠加：make_tracked_session(retry=Retry(total=0))。
在 429 时优先使用服务端提供的限流重置头——传输层已经遵循 Retry-After。让任何自定义重试保持有界且确定（stop_after_attempt），并有明确的终止行为。
将超时/重试设置放在模块顶部附近，方便调优。
未指定超时的请求会采用 default_request_timeout() 返回的默认 (连接, 读取) 超时（30 秒，300 秒）。当某个端点需要不同值时，在客户端配置中传递 request_timeout，或在 session 调用上传 timeout=。None 表示“使用默认值”。只有 NO_REQUEST_TIMEOUT 会发送没有截止时间的请求。

当客户拥有凭据时——他们自己在第三方账户上的 PAT / API key / OAuth token，几乎所有数据源都是如此——上述退避是合适的控制手段。 PostHog 无法超支它并不拥有的预算，因此数据源侧遵循 429 / Retry-After 就足够了。

例外是 PostHog 拥有并在多个进程间共享的凭据——目前就是 PostHog GitHub App 安装 token（许多 PostHog 子系统同时从每个安装的同一个预算中取用）。此时被动退避不够：如果没有协调，并发的 PostHog 调用方会在任何 429 返回之前集体突破共享上限。这些调用必须经由 posthog/egress/ 路由——一个由 Redis 支撑的共享预算加上遥测，并通过构造进行约束——绝不能手写 requests。在添加此类调用之前，先调用 /routing-outbound-api-calls。 GitHub 数据源是参考实现：它以 GitHub App 安装 id（GitHub 自身 id 空间中的预算所有者，而非 PostHog 数据库行）作为限流器的键，而客户 PAT 路径则在不感知 token 的情况下跳过限流器。对 api.github.com 的原始调用会被 github-api-calls-go-through-egress semgrep 规则阻止，因此 GitHub 形态的数据源通过构造就会落在 egress 路径上。决定性问题从来不是“这是不是数据仓库数据源？”——而是**“谁拥有 token，并发的 PostHog 进程是否可能因此互相踩踏？”**

扇出端点

扇出 = 遍历父资源，然后按每个父资源查询子端点。

单跳扇出优先使用依赖资源。 使用 rest_api_resources，父资源与子资源搭配，子资源为父字段声明 type: "resolve"。共享基础设施（rest_source/__init__.py、config_setup.process_parent_data_item）会对父资源分页，并按每个父行调用子资源。使用 include_from_parent 让子行携带父字段（通过 make_parent_key_name 注入为 _<parent>_<field>）。

让扇出声明式化。 在 settings.py 中添加扇出配置对象（例如 DependentEndpointConfig），包含 parent_name、resolve_param、resolve_field、include_from_parent、可选的父字段重命名以及可选的父端点参数。通过共享辅助函数（例如 common/rest_source/fanout.py:build_dependent_resource）来路由单跳扇出。

父字段重命名映射应放在辅助函数中。 调用方不应根据重命名是否存在进行分支判断。

按端点的分页/选择器 — build_dependent_resource 支持端点覆盖（parent_endpoint_extra、用于 paginator / data_selector 的 child_endpoint_extra、用于非 limit 大小参数的 page_size_param）。

路径预格式化： process_parent_data_item 只使用解析后的参数执行 str.format()。在传给资源配置之前，先用 .replace() 预格式化静态占位符，这样只会保留解析后的占位符。

仅当扇出深度为 2 层及以上时才使用自定义迭代器。 复用与其他地方相同的分页/重试辅助函数。

从数仓读取扇出父资源（`parent_source="warehouse"`）

重要

尚未普遍可用 — 不要将新数据源接入。 该标志仅限于内部试用推广，唯一已接线的调用方是 Sentry 的 issue_tag_values 自定义迭代器，并以增量水位线为条件。本节之所以存在，是因为共享扇出构建器无论如何都承载了这套机制，因此任何修改扇出内部实现的人都需要了解这些约束。请将以下注意事项（窗口化父资源、有序父资源、非 REST 调用方）视为在 Sentry 之外未经证实，并在将其他数据源接入之前咨询数仓团队。

默认情况下，扇出子资源会在每次同步时重新获取其父端点 — 同步 issue_hashes 会重新拉取所有 issues，即使 issues schema 已经同步过。子端点通过在其 DependentEndpointConfig 上设置 parent_source="warehouse" 来选择复用数仓父资源：随后子资源会从父 schema 已同步的 Delta 表流式读取父行（common/rest_source/warehouse_parent.py 中的 iter_parent_pages_from_warehouse），而不是访问父 API。参考实现：Sentry 的 issue_tag_values 自定义迭代器直接调用该读取器，仅当增量水位线限定扫描范围时才使用数仓模式。

要求与行为：

父级必须是同一来源中可选择的 schema —— 它必须生成自己的 Delta 表。
软依赖 —— 子级回退到父级 API。 在来源上重写 get_required_parent_schemas 来声明父级（将其接到 required_parents_from_endpoint_configs(ENDPOINTS, schema_name)；为自定义迭代器端点添加显式条目）。该重写是唯一的声明方式：没有任何东西会通过 API 暴露这层关系，因此在功能未经验证前，不要为它添加 schema 载荷字段。 API 中也没有任何东西约束选择：子级可以在没有父级的情况下启用，父级也可以在子级同步时被禁用或删除。import_data_activity_sync 中的 _warehouse_parent_reuse_available 按每次运行决定 —— 父级缺失、被禁用、尚未完成首次同步，或处于 merge 与 full refresh 之外的任何同步类型时，都会让该次运行走旧的父级 API 路径，因此让子级启用选择永远不会破坏一个当前能正常同步的 schema。仅仅处于同步中途的父级不会强制回退，因为 resolve_parent_table_ref 会通过 Delta 时间旅行将读取固定到父级最近一次完成的快照。绝不要因为启用子级而顺带启用父级：父级同步会计入客户的计费行数。
小父级留在 API 上。 低于 MIN_WAREHOUSE_PARENT_ROWS（1,000 行）的父级不值得开启：Delta 读取有固定几秒的开销，比翻页一个小列表还多，而且该群体在转换后实测更慢。该门槛读取父表的 row_count，因此无需配置即可按每次运行生效。不再有任何功能开关；回滚就是一次 revert。
严格流式 —— 绝不物化父表。 读取器一次扫描一个投影批次，并将列投影下推到 parquet 读取。不要为它添加 to_table、全局排序或已见集合去重 —— 父级可以任意大，而整条管线的存在就是为了避免全数据集驻留内存。如果调用方的语义依赖父级顺序（API 返回的是已排序的行），请将其改造成对无序流的逐行过滤（参见 Sentry 的 issue_tag_values 截止值处理），而不是排序。
可用的同步类型是允许列表，而不是拒绝列表。 只有 merge 和 full refresh 对每个键保留一行；append 每次同步累积一行，CDC 保留变更历史，因此对流式处理其中任何一种都会让子级按每个重复项扇出一次，而去重将需要无界状态。新的同步类型必须在 _parent_unusable_reason 中刻意选择加入。
值携带的是 Delta 物理类型，而不是 API 的 JSON 类型。 时间戳返回的是 datetime 而不是 ISO 字符串，嵌套对象返回的是 dict。由于 API 回退按每次运行启用，通过 include_from_parent 投影这样的字段会让子级的列类型在多次运行之间来回翻转，并触发 merge 的类型漂移防护。只投影物理类型与 API 返回值匹配的字段（id 字符串是安全的），或者在调用方进行规范化。
过期的父级会 404。 仓库快照可能包含自父级上次同步以来在上游被删除的父级；构建器会在子级上添加 404 → ignore 响应动作（自定义迭代器必须自行跳过 404 —— 参见 Sentry 的 _skip_rows_on_stale_issue_404）。
新鲜度：子级在父级最近同步的快照上扇出。在父级上次同步之后创建的父级会在父级重新同步后出现 —— 与独立调度属于同一类陈旧问题。
列名：读取器接收 API 字段名（例如 lastSeen），将其映射为 snake_case 的物理 Delta 列，并把行重新键回 API 名称。只请求扇出所需的列（resolve_field + include_from_parent）。
仓库读取必须复现 API 路径的有效行集，如果做不到，该父级就不合格。 供应商的列表端点通常在服务端限制了其返回内容，而增量父表会累积曾经见过的每一行 —— 无界扫描会在 API 路径绝不会涉及的父级上扇出，从而在陈旧快照上成倍放大子级行数、计费量和运行时间。在启用父级之前，将其归入以下三种情况之一：
- 父级 API 确实返回完整集合 → 无需过滤；理想候选。
- 该界限可知且可表达 → 在 DependentEndpointConfig 上设置 parent_row_filter（ParentRowFilter(field=..., not_older_than=..., not_before=...)）；该谓词会被下推到 parquet 读取，保留 NULL 行，适配字符串或时间戳物理列，而无法过滤的表会通过急切 resolve 校验回退到 API 路径。按每次运行的水位线（not_before）是最安全的界限，因为它与子级已处理的内容自洽。
- 该界限取决于你无法得知的状态 → 不要让该父级启用选择。 典型形态是受客户套餐限制的列表（事件保留期、席位层级、功能权益）：按账户、在服务端静默应用、没有任何 API 暴露，并且会覆盖你发送的任何显式范围。没有任何快照过滤器能复现这样的界限。有效界限通常在我们的代码中不可见，因此要通过实证来分类：用真实账户在带与不带显式界限的情况下运行该列表，比较数量，并检查疑似界限之外的项目是否仍能为它们的子级端点提供服务。
窗口化的父级必须保持窗口化。 如果来源目前按子级水位线限制其父级遍历（例如 Github 的 _fan_out_get_rows），那么完整的仓库读取会增加子级扇出 —— 应改为将仓库读取过滤到同一窗口。
非 REST 来源（例如 Stripe 的 SDK 循环）可以直接调用 iter_parent_pages_from_warehouse；投影其跳行检查所检查的任何字段（例如客户 balance）。在 source_for_pipeline（同步上下文）中急切地用 resolve_parent_table_ref(..., required_columns=[...]) 解析该表 —— 它会执行一次 ORM 读取，而管线的迭代器执行器线程并不适合临时数据库连接。它还会固定父级的 Delta 版本，因此一个在扇出中途重新同步的父级无法在子级底下移动行；将返回的 ref 传给读取器，而不是重新推导 URI。这样的来源还会跳过共享扇出的遥测，因此在其父级遍历结束时调用 log_fanout_parent_rows_consumed（sources/common/fanout_telemetry.py），并传入该次运行实际读取的 parent_source。没有那一行，就没有任何东西能将仓库扇出的规模与 API 路径的规模进行比较，而这是检测父级缺失列表仍返回的行的唯一手段。
围绕该 resolve 调用捕获 WarehouseParentTableNotFoundError 并走 API 路径。 一行 schema 记录可以声称同步已完成，而它的表却不可读（被清除、被重命名，或缺少扇出列），并且读取器是一个生成器，因此它懒加载校验的任何内容都会在管线深处抛出，那时已无处回退。这就是为什么 required_columns 在解析期间被急切校验。回退时，还要关闭任何只对仓库快照有意义的行为（子级的过期父级 404 → ignore，以及仓库扫描会跳过的续传检查点），使该次运行与功能关闭路径完全一致。

OAuth 配置

在实现 OAuth 之前，先检查该集成是否已存在——在 posthog/models/integration/ 包中按服务名粗略搜索一下，再断定它是新的。各类 kind 位于 model.py，OAuth 接线位于 oauth.py；只有当某个提供商在 OAuth 配置之外还带有业务逻辑时，它才会拥有自己的模块，Slack、GitHub 和 Stripe 就是如此。 __init__.py 只重新导出公开接口，因此仍然从 posthog.models.integration 导入，但要在定义它的模块中进行修改。

如果是新的：

环境变量。添加到 posthog/settings/integrations.py：

YOUR_SOURCE_CLIENT_ID = get_from_env("YOUR_SOURCE_CLIENT_ID", "")
YOUR_SOURCE_CLIENT_SECRET = get_from_env("YOUR_SOURCE_CLIENT_SECRET", "")

集成 kind。
- 添加到 posthog/models/integration/model.py 中的 IntegrationKind 枚举。
- 添加到 posthog/models/integration/oauth.py 中的 OauthIntegration.supported_kinds。
- 在 oauth_config_for_kind() 中添加一个 elif kind == "your-source": return OauthConfig(...) 分支，同样在 oauth.py 中。当环境变量为空时抛出 NotImplementedError("<Source> app not configured")——这就是失败即关闭（fail-closed）的消息，因此代码和 charts 可以在密钥值存在之前就先发布。
- 如果提供商的令牌响应中没有账户标识符（例如 Resend），则解码 access-token JWT，并从某个 claim（sub）设置 id_path / name_path，仿照 integration_from_oauth_response 中的 reddit/bing 分支。
注册客户端 + 部署凭据。 向提供商注册 OAuth 客户端、重定向 URI（US/EU/dev/localhost）、charts PR（将环境变量接入 Web 应用的 posthog-django-shared-secrets 和 worker 的 secret_env_app_specific 存储），以及通过 PostHog/secrets UI 或 CLI 将值写入 AWS Secrets Manager——另外还有其中哪些 Agent 可以自动化、哪些绝不能自动化——都见 references/oauth-app-deployment.md。

不可重试错误

重写 get_non_retryable_errors() 来标记那些应当永久失败而不是重试的错误：

def get_non_retryable_errors(self) -> dict[str, str | None]:
    return {
        "401 Client Error: Unauthorized for url: https://api.example.com": "Your API key is invalid or expired. Please generate a new key and reconnect.",
        "403 Client Error: Forbidden for url: https://api.example.com": "Your API key does not have the required permissions. Please check the key permissions and try again.",
    }

常见情况：401 Unauthorized、403 Forbidden、无效/过期的令牌、需要重新认证的 OAuth 令牌。

`validate_credentials`

在 source 创建时以 schema_name=None 调用（一次轻量探测，以确认令牌是真实的），并在每个 schema 的 incremental_fields 操作中以 schema_name=<name> 调用（确认该特定端点的 scope）。

如果 API 区分 401（令牌错误）和 403（令牌有效，缺少 scope），在 source 创建时接受 403——用户可能合理地只为其想要同步的端点授予 scope。仅当设置了 schema_name 时才重新抛出 403。同步时的 403 由 get_non_retryable_errors() 单独处理。

要在 schema 选择器中显示每个表的 scope 状态，请重写 get_endpoint_permissions(config, team_id, endpoints) -> {name: None | reason}：探测每个端点，当可访问时返回 None，否则返回一个简短原因。database_schema 操作会将其作为每个表的 permission_error 呈现出来，这样用户就能看到哪些表需要额外的 scope 并将其取消选择——它绝不能阻断 source 创建。基础默认实现将所有内容都报告为可访问。

当你确实要呈现缺失的 scope 时，请指明它——提供商通常会说明它（Required access: `read_x` access scope.），因此将其解析为你自己的消息，而不是直接抛出原始异常或将其简化为一个光秃秃的表列表。探测同步查询所需的任何字段（而不仅仅是 id），以便每个表的检查能反映同步该表实际需要什么。保持探测范围狭窄：只有真正的拒绝才是缺失 scope——限流、5xx 或网络抖动不是，因此将这些走可重试路径，而不是把每个异常都归类为“缺少权限”。

声明所需的 token scope

如果 API 会签发 OAuth scope 或按资源划分的访问 token，请声明源实际调用的每一个 scope，让用户清楚需要授予哪些权限——不要为了保险起见让他们授予全部 scope。

OAuth 源： 在 SourceFieldOauthConfig 上设置 requiredScopes（空格分隔的字符串，与 OAuth scope 参数格式一致）。前端会将其与集成已授予的 scope 进行比对，一旦有缺失，就向用户显示警告并提供 Reconnect 操作。
非 OAuth 源（PAT、API key）： 没有可检查的集成对象，因此改为在 caption 中列出 scope。Caption 通过 LemonMarkdown 渲染，因此反引号、粗体和链接均可用。

连接主机字段（凭据重定向）

如果你的源存储了密钥（API token、密码）并将其发送到用户在非 host 字段中配置的主机，请在该源类的 connection_host_fields 属性上声明该字段（继承自 common/base.py 中的基础源）：

@property
def connection_host_fields(self) -> list[str]:
    # `okta_domain` is where the stored API token is sent; retargeting it must re-require the token.
    return ["okta_domain"]

更新序列化器会读取该列表，并在其中任一字段发生变化时，强制编辑者重新输入该源的密钥。若没有它，组织成员就可能通过 PATCH 将主机字段改为自己控制的服务器，同时仍复用被保留（省略）的密钥——从而窃取该凭据。host 和 SSH 隧道目标已单独处理，因此只有连接目标位于不同名称字段中的源（例如 Okta 的 okta_domain）才需要覆盖此项。默认值为 []（没有额外字段）。

在创建源时，将其与 is_database_host_valid（common/mixins.py）配合使用，以拦截解析到内部/私有 IP 的主机。在连接时，通过 with_ssh_tunnel() 打开每个直接数据库连接，它会在每次打开时重新检查主机，并在驱动允许的情况下在客户端中固定已验证的地址（参见 common/mixins.py 中的 pinned_host_kwargs）。

Mixin

来自 products/warehouse_sources/backend/temporal/data_imports/sources/common/mixins.py：

SSHTunnelMixin —— 提供 with_ssh_tunnel() 上下文以及用于延迟打开隧道的 make_ssh_tunnel_func()。两者都会在每次打开时重新检查 SSH 主机和数据库主机。
OAuthMixin —— 通过 get_oauth_integration() 从数据库拉取 Integration。
ValidateDatabaseHostMixin —— 通过 is_database_host_valid() 拦截内部 VPC IP（除非使用了 SSH 隧道）。

图标

PNG 是仓库的惯例（source.template 默认使用 .png；约 800 个 png 对约 58 个 svg）。SVG 可以接受，但不是常规做法。请保持文件大小合理。
放在 frontend/public/services/ 中，并在 iconPath 中引用为 /static/services/{name}.png（或 .svg）——与提交的扩展名保持一致。
如果源 logo 尚不在项目中，请通过 Logo.dev 获取。向用户索取 API key——不要硬编码。如果用户未提供，请将其作为阻塞项提出，而不是提交占位符。

测试预期

在编写源测试之前，请先阅读 references/testing.md。 源测试约占 monorepo 测试用例的五分之一。 #113325 删除了其中大部分，因为按测试统计的覆盖率显示它们没有捕获任何问题。

准入门槛：在每个测试之前，说明它能捕获而其他任何测试都无法捕获的运行时故障。如果你说不出来，就不要写它。关于通用准入门槛，请参见 /writing-tests。

不要编写：

声明读取回验： lists_tables_without_credentials、connection_host_fields、api_docs_url、版本、get_source_config 类别、发布状态、字段或密钥标志、端点名称、主键、分页大小。sources/tests/ 中的共享不变量已经检查了每个源都关心的那些内容。
共享框架行为： get_non_retryable_errors() 的键成员资格、get_schemas(names=...) 过滤、get_documented_tables()、仅做转发的 source_for_pipeline，以及 rest_source 的分页器、认证和重试。
重复覆盖： 与多页遍历并列的第一页或单页测试、额外的恢复状态变体，以及针对共享同一代码路径的状态或模式分别编写的测试。请改为参数化。

在 tests/test_<source>.py 中编写：遍历到终止页的多页遍历、增量与全量刷新的请求构造、源在自行排序或窗口化行时的产出顺序、通过传输层由真实供应商响应驱动的错误映射、从保存状态恢复，以及边界情况输入下的行构造。仅当源类自身走的分支才添加 tests/test_<source>_source.py。一个典型的 REST 源需要一个包含 10 到 25 个参数化用例的传输模块。

实施检查清单

Bootstrapping:
- [ ] Enum added to products/warehouse_sources/backend/types.py (ALL_CAPS, no underscores between words) — the only place a source type is declared
- [ ] Source imported in products/warehouse_sources/backend/temporal/data_imports/sources/__init__.py + __all__
- [ ] Class inherits from SimpleSource / ResumableSource / WebhookSource (or combo) — see "Picking the right base class"

Source implementation:
- [ ] Set category on get_source_config (required — DataWarehouseSourceCategory; groups the source in the wizard catalog)
- [ ] Add keywords if the source has a common acronym / alternate spelling (optional, lowercase)
- [ ] Set api_docs_url (https, vendor API docs/changelog); add supported_versions + default_version if the vendor
      exposes a real version token — pin what the code actually calls (see "Vendor API version metadata")
- [ ] Define source fields in get_source_config
- [ ] Implement validate_credentials
- [ ] Implement get_schemas
- [ ] Add endpoint settings (settings.py)
- [ ] Implement transport + paginator ({source}.py)
- [ ] Return SourceResponse with correct primary_keys, partitioning, sort_mode
      (keys unique table-wide — parent id in fan-out child keys; sort_mode verified against actual response order;
      incremental cursor pagination stops at the watermark)
- [ ] Implement get_resumable_source_manager if ResumableSource
- [ ] Implement webhook methods if WebhookSource
- [ ] Add get_non_retryable_errors for auth/permission errors
- [ ] (Fixed-schema sources) Add canonical_descriptions.py from the API docs + override get_canonical_descriptions
- [ ] (Fixed-schema sources, static get_schemas only) Set lists_tables_without_credentials = True so public docs render the table catalog

Tooling & assets:
- [ ] Icon in frontend/public/services/ (PNG is the convention; ask user for Logo.dev key if needed)
- [ ] Run `pnpm run generate:source-configs`
- [ ] Swap generic Config for generated {Source}SourceConfig in source.py
- [ ] Run `hogli build:openapi`
- [ ] Django migrations run if enum value requires it

Release status (a finished source has NO unreleasedSource flag — it hides the source from users entirely):
- [ ] REQUIRED: delete `unreleasedSource=True` from the finished source (the scaffolded stub ships with it).
      Not being able to curl the live API is NOT a reason to keep it — use releaseStatus=ALPHA.
      Keep it ONLY when the code genuinely can't sync yet (landed across multiple PRs).
- [ ] No test asserts `unreleasedSource is True` (that anti-pattern locks the source hidden)
- [ ] When set, releaseStatus uses the `ReleaseStatus` enum, never a string literal
- [ ] releaseStatus=ReleaseStatus.ALPHA for a new source not yet extensively tested
      (ReleaseStatus.BETA later; ReleaseStatus.GA or omit for GA)
- [ ] featureFlag="dwh-{source_name}" ONLY if you want a controlled rollout instead of releasing to all

Tests & handoff:
- [ ] Tests follow references/testing.md: no declaration read-backs, no re-tested framework code, no duplicate coverage
- [ ] Source tests (test_<source>_source.py) — branches only, no declaration restatements
- [ ] Transport tests (test_<source>.py) — paginators, error mapping, request shaping
- [ ] User-facing doc written/updated per /documenting-warehouse-sources (docsUrl matches filename; `audit_source_docs` passes)
- [ ] `ruff check . --fix` and `ruff format .`
- [ ] List any new env vars (OAuth client IDs/secrets, etc) in the PR / handoff
- [ ] (Only if the source syncs an issues/tickets/conversations table) Note it as a Self-driving Inbox
      candidate — that's a separate follow-up via /adding-inbox-sources, not part of shipping the source

验证与生成工作流

更改源字段后，重新运行 pnpm run generate:source-configs 和 hogli build:openapi，然后运行新源的定向测试。对修改过的 Python 文件运行 ruff check . --fix 和 ruff format .。

常见陷阱

源在向导中不可见：未在 sources/__init__.py 中注册/导入，或未重新运行 hogli build:openapi。
test_source_categories 失败：源的 get_source_config 缺少 category —— 将其设置为最接近的 DataWarehouseSourceCategory 类别。
生成的配置类仍为空：更新字段后忘记运行 generate:source-configs。
增量同步行为异常：字段名/类型错误，或排序假设错误。
繁忙表上的 Pod OOM：主键实际并不唯一（通常是某个扇出子项在其键中缺少父 id）—— 重复行不断累积，每次合并都会对它们多重匹配；通常还伴随一个分页器，由于时间过滤器只作用于第一页，每次同步都会重新遍历完整历史。
在返回最新在前的 API 上声明了 sort_mode="asc"：第一批之后水位线就检查点到了 ≈当前时间，同步中途关闭会丢失数据顺序保证。
凭据无效却无限重试：缺少 get_non_retryable_errors。
token 有效但源无法连接：validate_credentials(schema_name=None) 会探测每个资源的 scope，而不只是 token，因此缺少一个 scope —— 通常是在用户不会同步的表上 —— 就会阻塞整个源。创建时只探测 token；通过 get_endpoint_permissions 报告每张表的 scope。
可恢复状态从未保存：忘记调用 save_state；或在一个覆盖了流水线尚未写入的行的游标上调用 commit()，这会在恢复时跳过这些行。
可恢复状态在其覆盖的 yield 之后才保存：一次交接会丢失最后一批的状态，因此下一次尝试会再次读取该批。posthog/test/repo_invariants/test_resume_state_staged_before_yield.py 会在循环中这样做的新的生成器上失败。其基线列出了现有的这些，且只能变短。
一次部署在可恢复源上等待数小时：它在没有行的响应中分页，且从不调用 manager.safe_point()，因此永远看不到 worker 关闭。
Webhook 行未落地：schema 的 is_webhook=False，或 initial_sync_complete=False。
依赖资源路径 KeyError：预格式化静态路径占位符（参见扇出）。
静默截断风险：达到页数上限却没有日志/指标。
重构导致的漂移：端点行为变化后遗留了未使用的函数参数/辅助函数。
端点配置字典中的类型漂移：使用源类型别名（Endpoint、ClientConfig、IncrementalConfig）以保持静态检查精确。
分区键不稳定：选择了 updated_at 而不是 created_at；每次同步都会重写分区。
提交了硬编码的 Logo.dev 密钥：始终在运行时向用户索取密钥。

支持 Markdown

登录后评论

基本信息

作者: posthog
仓库: posthog/posthog
分类: 数据与数据库
安装次数: 186
GitHub Stars: 40,496

报告问题