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/ にあるデータウェアハウスソースを構築または更新する際に使用します。

まず読むもの

コーディングの前に、以下を読んでください：

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 / DB プロトコル / webhook）およびトラッキング対象トランスポートの状態。まずこれをざっと読み、類似のソースがどのように配線されているか、また今回触るソースが現在どのような状態にあるかを把握してください。常に同期を保ってください — 後述の「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、フレームワークの認証とページネーター、トラッキング＋リトライ対応のトランスポート）を使用しており、レジューム可能です — 支配的な「リストエンドポイントをページネーションして、レジューム可能に yield する」形をフレームワークがカバーしていることの証明です。まずこれを、後述の「共有 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 をリファレンス実装として読んでください。

適切な基底クラスの選択

新しいソースはすべて、次のいずれか（またはその組み合わせ）を継承しなければなりません。

SimpleSource[Config] — 各実行がエンドポイントを完全に走査する、単純なプルベース API のデフォルトです。
ResumableSource[Config, ResumableData] — 基盤となる API が再開をサポートする、API ベースの新しいソースすべてで推奨されます（カーソル/Link ヘッダーによるページネーション、時間枠、オフセットトークン、その他中断した位置から確実に再開できるあらゆる手段）。API が次ページトークン、Link ヘッダー、または安定した時間フィルターを提供する場合は、ResumableSource を使用してください。これにより、Temporal はハートビートタイムアウト後に最初からやり直すことなく再開できます。save_state はカーソルをメモリ上にステージングし、パイプラインは、これまでに yield された行を書き込んだ直後にそれを Redis（TTL 24 時間）へコミットします。
WebhookSource[Config] — ソースがイベントをプッシュできる場合（例: Stripe の Webhook エンドポイント）にのみ使用します。通常は ResumableSource と組み合わせて、初回のバックフィルを再開可能にし、以降の差分は Webhook 経由で受け取ります。

両方が該当する場合は、多重継承で組み合わせます。例:

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

目安:

プルのみの API で、永続化できるカーソルがない → SimpleSource。
プルのみの API で、実行間に保存できるカーソル/次ページ/時間フィルターがある → ResumableSource。
ソースが変更イベントでコールバックできる → 適合するプル系の基底クラスに WebhookSource を追加。
ソースの次回実行が、カラムの最大値ではなくソースだけが計算できる位置から始まる → CursorSource を追加。

データベースおよびファイル転送ソース（SFTP、S3）は、明確な理由がない限り SimpleSource のままにします。

永続的なソースカーソル（`CursorSource`）

インクリメンタルフィールドは一般的なケースをカバーします。つまり、パイプラインがカラムの最大値を取得し、次回の実行でそれを超える値にフィルターします。一部の位置はカラムの最大値ではありません。読み取り前に取得した Postgres の xmin 上限や、Kafka のパーティションオフセットの集合などです。そのような場合は、SourceResponse に新しいフィールドを追加したり、sync_type_config に新しいキーを追加したりするのではなく、sources/common/cursor.py の CursorSource[CursorT] ミックスインを追加してください。

カーソルは、cursor_kind: ClassVar[str] を持つ @frozen データクラスとして定義します。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 では、ローダーが最終バッチとともにそれを昇格させます）、リセットはそれをクリアします。postgres/xmin_cursor.py がリファレンスです。

共有 REST フレームワークを優先する

ほとんどの REST ソースは、自作のクライアントではなく、共有の rest_source フレームワーク (common/rest_source/) 上に構築すべきです。このフレームワークはすでに以下を提供しているため、あなたが書く必要は まったく ありません:

追跡・リトライ対応のトランスポート — RESTClient はデフォルトで make_tracked_session() を使用し、 Retry-After を尊重しつつ 429 と一時的な 5xx をリトライします。tenacity も RetryableError もフェッチループも不要です。
ページネーター (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 (顧客所有のクライアントクレデンシャル/リフレッシュ)。それぞれが自身のシークレットを秘匿するため、_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() を利用し、2 つ目のステータスコードリトライ層を追加しないでください (「リトライとスロットリング」を参照)。

新しい API ソースのエンドツーエンドワークフロー

この順序に従ってください。各ステップは source.template の TODO に対応しています。

ソースを調査します。 ユーザーが実際に必要とするエンドポイントを選びます。以下を相互参照してください:
- Airbyte: https://airbyte.com/connectors (コネクタページはしばしばソースコードへリンクしています — 有用な参考情報です)
- Fivetran: https://www.fivetran.com/connectors
- Stitch: https://www.stitchdata.com/docs/integrations/ 公式 API ドキュメントまたは OpenAPI 仕様を見つけ、リクエストコードを書く前に、ベンダーの最新の一般提供 API バージョンを特定してください — それがソースを構築すべき対象のバージョンです。ベンダーのチェンジログ、バージョニング、または非推奨化ページを確認してください。検索で最初に出たページだけを見てはいけません。ドキュメントサイトは古いバージョンをデフォルトにしていることが多く、Airbyte/Fivetran コネクタはしばしば数年遅れています。「ベンダー API バージョンメタデータ」で、何が最新に該当するか、最新チャンネルが GA でない場合の対処を参照してください。
ソースをブートストラップします。 テンプレートをコピーし、enum/型の参照を配線します:
```
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
```
次に、enum メンバーを 1 つ追加します。products/warehouse_sources/backend/types.py にある ExternalDataSourceType (products/warehouse_sources/backend/facade/types.py から再エクスポート) が、ソース型を宣言する唯一の場所です — そのファイルの既存の慣例に従ってください: ALL_CAPS で単語間に アンダースコアなし (例: ACTIVECAMPAIGN、APPLESEARCHADS)、値は PascalCase です。(フロー内で唯一のケバブケース識別子は、オプションの featureFlag="dwh-{source_name}" です。)
ベースクラスを選択（上記を参照）し、クラス名／source_type の戻り値を変更します。
get_source_config を定義します。name、category（必須。「ソースのカテゴリとキーワード」を参照）、label、caption、docsUrl、iconPath、fields、任意の keywords を設定します。適切なフィールド型を使用してください（以下を参照）。ベンダー API バージョンのメタデータクラス属性も設定します。「ベンダー API バージョンのメタデータ」を参照してください。ソースのリリース時には、「コードベースの検出」の説明に従って codebase_detection も設定します。
ソースを登録します。products/warehouse_sources/backend/temporal/data_imports/sources/__init__.py に import 行を追加し、__all__ に含めます。（クラスの @SourceRegistry.register デコレーターが実行時の登録を処理します。）
設定ジェネレーターを実行します。pnpm run generate:source-configs を実行してください。新しい設定クラスが products/warehouse_sources/backend/temporal/data_imports/sources/generated_configs/<your_source>.py に生成されていることを確認します（ソースごとに生成モジュールが 1 つ作成されます。パッケージの __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 に対して SVG は約 58 個で、source.template のデフォルトも .png です）。SVG も使用できますが、標準ではありません。コミットする拡張子に合わせて iconPath を設定してください。ロゴがまだコミットされていない場合は、Logo.dev から取得します。Logo.dev の API キーはユーザーに確認してください。ハードコードしないでください。Logo.dev の画像 API が返すのは SVG ではなく PNG です。ファイルサイズは適切に抑えてください。
マイグレーションを実行します。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 にすると、コネクターがユーザーから完全に非表示になります。フロントエンドはこの値が truthy のすべてのソースを除外します。InlineSourceSetup.tsx と、nonHogFunctionTemplatesLogic.tsx の「近日公開／通知を受け取る」フローを参照してください。） その行を削除することは必須であり、あなたの環境でできないことによって条件付けられるものではありません。 特に、「ライブ API に curl できなかった」とか「実際のアカウントで検証できなかった」というのは、フラグを残す理由にはなりません。まさにそれが releaseStatus=ReleaseStatus.ALPHA の役割です（可視のソースに付ける、ソフトな「新規で、 lightly tested」というラベル）。unreleasedSource=True が正当に残るのは、そのソースが複数の PR にまたがって着地中で、実装コードがまだ全部揃っていないために物理的にまだ同期できない場合だけです。get_schemas / source_for_pipeline が動作し、テストも通っているソースは完成です。フラグは外します。unreleasedSource is True をアサートするテストは絶対に書かないでください — それはバグを固定してしまい、166 個の完成済みソースが一括リリースされるまで隠され続けた原因です。

したがって、新しく完成してテスト済みのソースは次のように出荷されます：
- unreleasedSource なし（可視かつ接続可能）、
- 広くテストされていない新しいソースには releaseStatus=ReleaseStatus.ALPHA（粗削りな部分が解消されたら ReleaseStatus.BETA、一般提供なら ReleaseStatus.GA、または releaseStatus を完全に省略） — 可視のソースに付けるソフトなラベルであり、ゲートではありません、
- オプションで featureFlag="dwh-{source_name}"（ケバブケース）を 付けるのは、全員にリリースするのではなく、フラグが有効なユーザーへ制御されたロールアウトを行いたい場合のみです。
releaseStatus を設定するときは常に、products.warehouse_sources.backend.facade.source_config の ReleaseStatus 列挙型を使用してください — 裸の文字列リテラルは決して使わないでください。そのモジュールからの既存の import ブロックに ReleaseStatus を追加してください。
ソースを文書化する。 /documenting-warehouse-sources スキル（テンプレート、共有スニペット、<SourceParameters /> + <SourceTables />）に従って、posthog.com のユーザー向けドキュメントを書くか更新してください。get_source_config 内の docsUrl がドキュメントのファイル名（ケバブケース）と一致することを確認し、さらに — get_schemas が静的なエンドポイントカタログである場合は — lists_tables_without_credentials = True を設定して（下記参照）、ドキュメントの Supported tables セクションがレンダリングされるようにしてください。完成したソースはスタブではなく、一貫したドキュメントとともに出荷されます。
PR の前にテンプレートの TODO コメントを削除してください。

ソースアーキテクチャ契約

API ベースのソースでは、次の分割を使用してください：

source.py: ソース登録、ソースフォームフィールド、スキーマ一覧、認証情報の検証、再開可能/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、カスタムマニフェスト） → ソースがオプトインするケイパビリティミックスインとし、API は isinstance(source, <Capability>) でディスパッチします。
直接クエリエンジンの動作（SQL エンジンがテーブルロケーションを解決する方法、DataWarehouseTable を構築する方法、カラムをマッピングする方法）は、ソースタイプではなくエンジンに基づいてキー付けされます — source_type ではなく、エンジンアダプター/レジストリ（クエリ関心事は posthog/hogql/direct_sql/、マテリアライゼーションは data_warehouse エンジンレジストリ）を通じて source.direct_engine でディスパッチします。

ソースドメインのセマンティクス（エンジンとの通信方法、名前の付け方、フィルターがプッシュダウンされるかどうか）はソース側に保ち、それが駆動するウェアハウスドメインの処理（DataWarehouseTable の行、マネージド ViewSet、Hog 関数）は data_warehouse に置き、ソースまたはアダプターが返すものに基づいてキー付けします。ソースのケイパビリティは data_warehouse の型を決してインポートしません。 products/data_warehouse/backend/presentation/README.md を参照してください。

トップレベルとファンアウトの両方のエンドポイントが混在する REST ソースでは、エンドポイントのメタデータを settings.py に保ち、{source}.py で次の優先順位でルーティングします。

エンドポイント固有のカスタムイテレーター（必要な場合のみ）、
汎用ファンアウトヘルパーパス、
トップレベルエンドポイントパス。

正規の説明（セマンティックエンリッチメント）

テーブルが同期された後、バックグラウンドアクティビティ（workflow_activities/enrich_table_semantics.py）が各テーブル/カラムを記述する WarehouseColumnAnnotation 行を書き込み、AI エージェントに提示されます。固定スキーマのソース（SaaS API）ではスキーマは誰にとっても同じであるため、チームごとに LLM に再導出させるコストを支払う代わりに、公式 API ドキュメントから一度だけ文書化します。これらのキュレーションされた説明は権威があり、直接適用され（description_source="canonical"）、LLM に送信されることはありません。

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).",
        },
    },
}

次に、ソースクラスのフックをオーバーライドし、兄弟ファイルを遅延インポートします。

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 が返すエンドポイント/スキーマ名（ENDPOINTS と一致）でキー付けします。
説明は推測ではなく、公式 API ドキュメントから取得します。部分的なカバレッジで問題ありません — 欠落しているエンドポイント、カラム、またはテーブルレベルの description は LLM にフォールバックし、LLM にはソース名、エンドポイント、docs_url、カラムのデータ型が渡されます。
任意であり、固定スキーマのソースに対してのみ意味があります。SQL ソース（任意のユーザースキーマ）は何も提供しません — ベースフックは {} を返します。
source.py/settings.py の転送ロジックには触れないでください — これは純粋に追加的なメタデータです。

テーブルカタログを公開ドキュメントに公開する

posthog.com のドキュメントは、public_source_configs API によって供給される <SourceTables /> コンポーネントを通じて Supported tables セクションをレンダリングします。この API は各ソースの get_documented_tables() を呼び出します。基本の実装では、get_schemas(および canonical_descriptions とマージされたもの)からテーブルを一覧表示しますが、それは ソースがオプトインした場合のみ です:

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

これを True に設定するのは、get_schemas が I/O を伴わない 静的なエンドポイントカタログを反復処理する場合のみです — ネットワークも、DB も、認証情報もなし(よくある固定スキーマの SaaS パターン: for endpoint in ENDPOINTS)。このエンドポイントはプレースホルダーの設定を構築し、実際の認証情報なしで get_schemas を呼び出します。そのため、スキーマを検出するために接続するソース(SQL、ファイルストレージ、MongoDB、アカウントを一覧表示する広告プラットフォーム)は、これを False(デフォルト)のままにしなければなりません — そうしないと、空のホストに接続しようとしたり、ハングしたり、DB セッションを閉じてしまったりします。False の場合、ドキュメントは代わりに「あなたのアカウントから検出されました」という汎用的な注記をレンダリングします。

テーブル一覧が豊富であるほど、ドキュメントは良くなります — そのため、これを canonical_descriptions.py(テーブル/カラムの説明)と組み合わせてください。レンダリングされた出力は API で確認できます: GET /api/public_source_configs → あなたのソース → tables。

ソースのカテゴリとキーワード

すべてのソースは、その SourceConfig に category を設定 しなければなりません — これは、新規ソースウィザードのカタログ(カテゴリレール + タイルグリッド)でソースをグループ化します。登録されたソースにカテゴリがない場合、テスト(tests/test_source_categories.py)が失敗するため、これは省略できません。SourceConfig と DataWarehouseSourceCategory を products.warehouse_sources.backend.facade.source_config から、name をファサードの ExternalDataSourceType からインポートします:

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"],
    ...
)

最も近いバケットを 1 つ選んでください。enum のメンバー(ラベルに " & " がある場合は三重アンダースコアであることに注意してください):

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 — メール/SMS/マーケティングオートメーション(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 がそれからフロントエンドの型を再生成します。新しい カテゴリを追加するということは、その enum を編集することを意味します — アドホックな文字列をでっち上げないでください。

keywords は小文字の検索エイリアスのオプションのリストです — ユーザーが入力する可能性のある一般的な頭字語や別の綴りがソースにある場合にのみ追加してください(例: ["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 は、リリース済みのソースにこの属性がない場合、および未リリースのソースにこの属性がある場合に失敗します。イメージのビルド時に宣言がウィザードのダウンロードするカタログに反映されるため、更新が必要な生成ファイルはありません。

Self-driving Inbox の候補（イシュー／チケット／会話）

一部のソースは、Self-driving Inbox の候補でもあります。これは、同期された 対応が必要なレコード のテーブルを監視し、PostHog Desktop Inbox に検出結果を出力する機能です。現在提供中の対象は、GitHub、 Linear、Zendesk、pganalyze、Jira です。

シグナルとなるのはベンダーではなく、同期するテーブルです。テーブルのひとつが、人間（またはエージェント）が一件ずつトリアージするレコードのストリーム、つまり issues、tickets、conversations テーブルである場合、そのソースは Inbox の候補です。これらは、サポート／ヘルプデスク（CUSTOMER_SUPPORT）、イシュートラッカーおよびモニタリング（ENGINEERING___MONITORING）、一部のプロジェクト管理ツール（PRODUCTIVITY）のカテゴリに含まれます。分析、請求、広告プラットフォーム、CRM、生のデータベースソースは Inbox の候補ではありません。同期するのはクエリ対象のデータであり、対応すべき作業キューではないためです。作成中のソースにこのようなテーブルがない場合、ここで行うことはありません。

ソースを Inbox に接続する作業は、独自のスキル /adding-inbox-sources を使う別個の追加作業であり、このスキルの成果物には何も変更を加えません。この作業はデータウェアハウスのソースが存在して初めて可能になります（このスキルが作成するのがまさにそのソースです）。そのため、まずソースを構築してリリースしてください。このスキルでは、3 つの領域を変更します。このリポジトリ内のサーバー側「signals scout」エミッター、レジストリエントリー、SignalSourceProduct enum（products/signals/backend/）、別リポジトリ posthog/code の Inbox UI、そして PostHog/context-mill の npx @posthog/wizard self-driving オンボーディングフローです。開始前に /adding-inbox-sources を読んでください。これらの仕組みは、ソース独自の products/warehouse_sources/ コードには含まれません。

ベンダー API バージョンメタデータ

すべてのソースは、ベンダーの API バージョンを記述する 3 つのクラス属性を(ソースクラス本体に、lists_tables_without_credentials と並べて)宣言します。フレームワーク(common/base.py)は、各 ExternalDataSource がどのバージョンに対して実行されるかを記録するため、古いピンが引き続き機能し、非推奨化を可視化できます。 sources/tests/test_source_versions.py は、登録されたすべてのソースに対して以下の不変条件を強制するため、これらを誤った新しいソースは CI で失敗します。

2 つのケースがあります:

ベンダーが実在するピン留め可能な API バージョンを公開している場合 — URL パスセグメント(/v3/、/2/)、必須のバージョンヘッダー値(日付形式の 2022-11-28)、日付付きのクエリ/バージョンパラメータ、または名前付きリリース。3 つすべてを宣言します:
```
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)
```
ソースはベンダーの最新の一般提供(GA)バージョンに対して構築し、それをピン留めしてください。 新しいソースは 1 つのバージョンで開始し、それを接続するすべての顧客がそこに到達するため、古いバージョンで出荷することは、後で誰かが実行しなければならないマイグレーションを出荷することを意味します。2 つのルールがあり、それらは一致していなければなりません:
1. ベンダーが提供する最新の GA バージョンに対してリクエストコードを記述します。
2. そのコードが実際に呼び出すバージョン(ベース URL パス、バージョンヘッダー、または settings.py / {source}.py 内のバージョン定数)を宣言します。コードが送信しないバージョンを宣言してはいけません — そのピンはフレームワークが検出できない嘘であり、すべての顧客にとって非推奨警告とアップグレードパスを誤ったものにします。
最新バージョンに到達できない場合 — それがプレビュー/ベータ/不安定/RC である、申請や有料プランの背後に制限されている、またはそのレスポンス形状がまだ実装されていない — 実際に呼び出せる最新の GA バージョンに対して構築し、それをピン留めし、その理由をクラスのコメントに記述してください。「最新」とは最新の安定版を意味します: Shopify の unstable、ベンダーの -rc チャネル、またはドキュメントがまだプレビューとマークされているバージョンをピン留めしないでください。

ツリーに既にある例: Anthropic ("2023-06-01",)(日付付き anthropic-version ヘッダー)、ActiveCampaign ("v3",)(/api/3 パスセグメント)、Alguna ("2026-04-01",)(日付付きバージョンヘッダー)。後で 2 つ目のバージョンを獲得するソースは、古い順から新しい順に宣言します — 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:// でなければなりません。
ベンダーの正確なバージョン文字列を使用してください。決してでっち上げないでください。
ベンダーがすでに非推奨とした、またはサポート終了日を設定したバージョンで新しいソースを出荷しないでください。 唯一のバージョンをカバーする deprecated_versions エントリを持つ真新しいソースはバグです — それはソースが誤ったバージョンに対して書かれたことを意味します。default_version が非推奨の場合、test_source_versions.py はビルドを失敗させます。
api_docs_url は、一般的な API ランディングページよりも、ベンダーのバージョニング/変更履歴ページを指すものを優先してください — そこが次のバージョンが発表される場所であり、次の担当者が再ピン留めの前に確認する場所です。
トランスポート/リクエスト層でフォールバックバージョンをハードコードしないでください — ソースクラスから解決してください(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 として union されます。

SourceFieldInputConfig — 基本的な入力（text、email、number、password、textarea）。<LemonInput /> としてレンダリングされます。
SourceFieldSwitchGroupConfig — フィールドのサブグループを表示するトグル。オプションの機能ブロックに使用します。
SourceFieldSelectConfig — ドロップダウン。オプションは選択時に表示されるサブ fields を持てます（代替の認証方法に使用します — 例: API キーと OAuth）。
SourceFieldOauthConfig — Integration モデルを介した OAuth。OAuth のセクションを参照してください。
SourceFieldFileUploadConfig — ファイルアップロード（JSON）。keys=["..."] の許可リストまたは "*" を使用します。
SourceFieldSSHTunnelConfig — SSH トンネルのサブフィールドをレンダリングします。ヘルパーとともに ssh_tunnel: SSHTunnel を設定に追加します。

ガイドライン:

複数の認証方法 → オプションごとに子 fields を持つ SourceFieldSelectConfig。
オプションのトグル → SourceFieldSwitchGroupConfig。
機密フィールドには SourceFieldInputConfigType.PASSWORD を使用する必要があります。シリアライザーはフィールド定義から機密キーと非機密キーを自動的に導出します — 許可リストを別の場所で管理する必要はありません。

`source_for_pipeline` の実装

SourceResponse を直接返してください。新しいソースでは dlt_source_to_source_response を使用しないでください — DLT は削除される予定です。

API が返す形でデータを yield することを推奨します。カスタムデータクラスや重いパースは不要です。dict、list[dict]（可能な場合はこちらが推奨）、または pyarrow.Table のいずれかを yield してください。バッファリングとバッチ処理はパイプラインが行います。

デフォルトでは生の dict / list[dict] を yield し、バッチ処理はパイプラインに任せてください。 パイプラインはすでに Batcher（pipelines/pipeline_v3/pipeline.py）を 5000 行 / 200 MiB のしきい値で実行しているため、通常のケースでは独自のバッチャーは不要です。pyarrow.Table に頼るのは、すでに arrow 形式のデータがある場合（例: ClickHouse アダプター）だけにしてください。ソースが独自の Batcher をより小さいしきい値（例: chunk_size=2000, chunk_size_bytes=100 * 1024 * 1024。klaviyo や他の約 70 のソースがそうしています）でインスタンス化してよいのは、大きな/幅の広い行に対して意図的にメモリ使用量を抑えたい場合です — これは有効な選択であり、デフォルトではありません。避けるべきなのは 2 つ目のフルサイズのバッチャーで、これは利点のない二重バッファリングになるだけです。

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 安全なパラメータに対して同一の出力を生成します。

再開可能なソースパターン

@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)

トランスポート関数内では次のようにします。

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 の後に保存しないでください。ワーカーのシャットダウン時、パイプラインは制御がソースに戻る前に試行を終了するため、yield の後に保存された状態は最後のバッチについて失われます。次の試行はそのバッチを再度読み取り（マージは主キーで重複排除し、再開された完全リフレッシュはそれを二重に追加します）、1 バッチ以下しか書き込まない試行は進捗を保持しません。まだ何も yield していないソース、たとえばエクスポートジョブ ID をポーリング前に永続化するソースは、with manager.committing(): の内部でステージングし、ブロックの終了時にコミットします。

保存されたカーソルは、パイプラインがそれを確認した後にのみコミットできます。パイプラインは、ソースが次のアイテムを渡したとき、終了したとき、または安全なポイントに到達したときに確認します。したがって、最後の yield の後に保存されたカーソルは、ソースが例外を発生させたときには永続化されません。ソースがカーソルのスキップする行をまだ保持している可能性があるためです。次の試行は最後の yield のカーソルから続行します。ページまたは時間の予算で自身の試行を終了するソースは、ローカルバッファが空の状態で、例外を発生させる直前に manager.safe_point() を直接呼び出して、最後のカーソルを保持します。

ソースが行を返さない多数のリクエストを行う可能性がある場所では、manager.safe_point() を呼び出してください。たとえば、空のデルタページ、子を持たないファンアウト親、コメントのないページなどです。パイプラインはアイテムが到着したときにのみワーカーのシャットダウンをチェックするため、そうでなければ一連の空のレスポンスがワーカーをグレースフルシャットダウンのタイムアウト全体にわたって保持し、そのカーソルは決してコミットされません。安全なポイントでは、パイプラインは実行を別のワーカーに引き渡すことができ、書き込み待ちが何もないときにステージングされたカーソルをコミットします。これは、ステージングされたカーソルから再開しても行が失われない場所でのみ呼び出してください。カーソルが対象とするすべての行がすでに yield されており、ローカルバッファに残っていないことが条件です。参照: convex/convex.py の document_deltas、stripe/stripe.py のスパーススイープチェックポイント、notion/notion.py の _page_fan_out。 rest_source フレームワークは、各ページの後と各リトライ待機の前に、それ自身で安全なポイントに到達しますが、それは SourceResponse.items がフレームワークの Resource を直接返す場合に限られます。それをラップするソースはフレームワークの安全なポイントを得られません。ラッパーが行をバッファリングする可能性があるためです。同じ条件が resume_hook の実行タイミングを決定します。items が Resource を直接返す場合、フックはページがパイプラインに到達する前に実行されるため、ページとそのカーソルが一緒にコミットされ、ハンドオフで行が重複しません。ソースが Resource をラップする場合、フックはラッパーが次のページを要求したときに実行されるため、ハンドオフで最後のページが再度読み取られます。可能な場合は Resource を直接返してください。行の変更には data_map、add_map、add_filter を使用します。各ページを変更せずに引き渡し、行を保持しないラッパーは、Resource(wrapper, name=..., hints=resource._hints) を返すことでフレームワークの動作を維持できます（anthropic/anthropic.py の使用状況レポートを参照）。 resume_hook 内で safe_point() や commit() を呼び出さないでください。

ソースコントラクト

シャットダウンするワーカーは、実行中の各インポートを別のワーカーに引き渡します。抽出コードを持つすべてのソースは、それを可能にしなければなりません:

境界付き呼び出し: 各リクエストには読み取りタイムアウトがあり、リトライと待機を含む 1 つのリクエストがワーカーを占有するのは最大 5 分間で、これは大きな Retry-After を伴う 429 の後でも同じです。
各リクエストの安全点: ソースは、yield も安全点もなしに連続して 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 ペイロードを変換する HogFunctionTemplateDC を返す webhook_template を実装します。
当社のスキーマ名 → 外部オブジェクト型をマッピングする webhook_resource_map を実装します。
API がプログラムによる webhook 管理を許可している場合は、create_webhook、delete_webhook、get_external_webhook_info を実装します。そうでない場合は、失敗した結果を返し、手動セットアップを説明する webhookSetupCaption を提供します。
セットアップ後の入力 (例: 署名シークレット) のために SourceConfig に webhookFields を追加します。
source_for_pipeline 内で、self.get_webhook_source_manager(inputs) を呼び出し、そのイテレーターをプルイテレーターと並べて渡して、1 回の同期で履歴行 + webhook 配信行をプルできるようにします。
SourceSchema.supports_webhooks=True は、webhook が実際に実行可能なエンドポイント (通常はインクリメンタル/追記専用のもの) にのみ設定します。
webhook バッチ内では table_transformer で重複排除します。 WebhookSourceManager.get_items() は、生の webhook ペイロードが行ディクショナリにデシリアライズされた後に適用される、オプションの table_transformer: Callable[[pa.Table], pa.Table] を受け取ります。Delta マージは (primary_keys に基づいて) 同期を_またいで_のみ重複排除し、単一のソースバッチ内では行いません。したがって、1 つのバッチが同じオブジェクトに対して複数のイベント (例: 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"] でのマージに対応)。

マルチスキーマ SQL データベースソース

SQL DB ソース（現在は Postgres、MSSQL、Snowflake、Redshift）は、1 つの接続で全ネームスペース（スキーマ）のテーブルをインポートできます。ネームスペースフィールドを空にすると、すべての非システムネームスペースからテーブルが検出され、ウィザードはそれらをネームスペースごとにグループ化し、同期は namespace.table ごとに 1 つのウェアハウステーブルを書き込みます。リファレンス実装: postgres/postgres.py + PostgresImplementation。共通の接続面は common/sql/ にあります。

機能マーカーは、get_source_config においてソースの schema フィールドが オプション（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 フィールドで required=False とし、pnpm run generate:source-configs を再実行します。database は必須のままにしてください。データベース/カタログは接続ごとに固定です。
マルチネームスペース検出 — 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) を解決します: 行ごとのメタデータ → ドット名の自己修復 → 設定ネームスペースの順です。解決されたスキーマ + 非修飾テーブルに対して SQL を実行し、SourceResponse.name = response_name（dwh_storage_key or schema.name、正規化済み）を設定します — 裸のテーブル名にしてはいけません。そうしないと行の Delta パスが移動し、同期済みデータが孤立します。
解決されたネームスペースをすべてのストリーミング/統計ヘルパーに通す — テーブルメタデータ、行統計、平均行サイズ、パーティション設定、チャンクサイズ、主キー検索はすべて (schema, table) を受け取ります。1 つでも欠けると、静かに劣化します（パーティションなし / 誤った統計）。
ドット付き表示名を識別子クォータに渡さない — quote("a.b") は 1 つの誤った識別子を生成します。まず (schema, table) に分割し、quote_qualified（common/sql/identifiers.py）を使用してください。
命名レイヤーは派生であり、保存されない — 表示名 analytics.users、S3/Delta サブディレクトリ analytics_users（正規化された response_name）、HogQL テーブル {prefix}_analytics_users。唯一保存される例外は dwh_storage_key で、移行されたレガシー行を元の Delta パスに固定します。
レガシー移行は機能駆動であり、ソースタイプで制限されない — ユーザーが既存の単一スキーマソースでネームスペースをクリアすると、sql_warehouse_migration.py が行を修飾形式にリネームし、dwh_storage_key を付与して、再同期なしで同期済みデータを保持します。共有レイヤーに source_type == "..." の分岐を追加しないでください。
テスト — 同じテーブル名を持つ 2 つのネームスペースがエンドツーエンドで区別されたままであること。空のネームスペースの検出がシステムネームスペースを除外すること。行ごとのルーティングが正しいネームスペースに到達すること。レガシーな単一ネームスペースソースが引き続き動作すること。移行された行がレガシー Delta パスを維持すること。

検出コスト: validate_credentials と database_schema は名前フィルターなしで検出を実行するため、数百のスキーマを持つカタログでネームスペースを空にしても、ネームスペースごとにテーブルごとのクエリを発行してはいけません — リストクエリをバッチ化するか、列挙を上限付きにしてください（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 から）を経由しなければなりません。追跡対象のセッションは、すべての送信リクエストのログ行と OTel メトリクスに team_id、source_type、external_data_source_id、external_data_schema_id、external_data_job_id を付与し、オプトインのサンプルキャプチャに参加します。

生の requests を使用する場合: make_tracked_session(headers=..., retry=...) は requests.Session を返します。モジュールレベルの requests.get/... ショートカットの代わりに session.get/post/... を使用してください。
キャプチャされたサンプルからシークレットを秘匿してください。 make_tracked_session(...) に redact_values=(api_key, token, ...) を渡すことで、追跡対象のトランスポートがログに記録された URL、ヘッダー、サンプリングされたボディ内のこれらのリテラル値をマスクします。これはクエリパラメータや特殊なヘッダー名に含まれるキーにとって重要です。rest_source の認証クラスはこれを自動的に行います（それぞれが secret_values() を実装しています）。生のセッションを使用するソースのみ、自分で redact_values を渡す必要があります。
すでに rest_source.RESTClient を経由しているソースの場合: 自動的に追跡対象のセッションがデフォルトになるため、変更は不要です。
セッション/HTTP クライアントのフックを受け入れるベンダー SDK（Stripe の RequestsClient(session=...)、gspread の authorize(credentials, session=...)、AuthorizedSession + TrackedHTTPAdapter 経由の BigQuery）の場合: 1 つ注入してください。参考パターンは stripe/stripe.py、google_sheets/google_sheets.py、bigquery/bigquery.py にあります。
注入ポイントのないベンダー SDK（現時点では bingads、linkedin-api の RestliClient）の場合: 1 行の理由を添えた # nosemgrep: data-imports-http-transport-... プラグマを追加し、SOURCES.md でそのソースを ⚠️ Vendor SDK として記録してください。
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）にデータを供給します。2 つの接続ポイントがあります:

interceptors= リストを受け入れる SDK（google-ads の GoogleAdsClient.get_service(...)）の場合: すべての get_service 呼び出しで interceptors=tracked_interceptors(host) を渡してください — google-ads は呼び出しごとにチャネルを再構築するため、インターセプターは毎回再指定する必要があります。参考: google_ads/google_ads.py。
channel= / transport= を受け入れる SDK（BigQuery Storage Read API）の場合: 認証情報を含むチャネルを構築し、make_tracked_channel(channel, host=...) でラップしてから、トランスポートに渡してください。参考: bigquery/bigquery.py:bigquery_storage_read_client。

CI はこれを .semgrep/rules/security/data-imports-grpc-transport.yaml によって強制します。このルールは sources/** 内（common/grpc/ パッケージと 2 つの参考ソースファイルの外側）での生の 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 テーブルに移動します。
以前に scaffold されたソースを実装する — 行を Implemented テーブルに移動し、通信方法、主要ライブラリ、追跡トランスポートの状態を記入します。
ベンダー SDK を移行して追跡対象のセッションを注入する — ソースを ⚠️ Vendor SDK から ✅ に切り替えます。
ソースのプロトコルを切り替える — たとえば REST を gRPC に置き換える、プル API と併用して webhook サポートを追加する、requests からベンダー SDK に移行するなど。通信方法と追跡トランスポートの両方の列を更新します。

各テーブル内ではエントリをアルファベット順に保ってください。scaffolded リストは 1 ソースにつき 1 行(1 つの箇条書き、これもアルファベット順)であるため、ソースの追加や削除はその行だけに影響し、同時進行の PR との競合を避けられます — カンマ区切りの段落に戻さないでください。部分的に追跡されているソースを追加する場合は、追跡を妨げているもの(たとえばセッション/インターセプターの接続点がないベンダー SDK など)を説明する短い「部分的に追跡されているソースに関する注記」エントリも追記してください。

基底クラスの機能フラグと API バージョニング

common/base.py は、ほとんどの API ソースがデフォルトのままにしているクラスレベルのフラグを公開していますが、該当する場合には重要になります:

supports_column_selection(デフォルトは False、SQLSource は True を設定)— ソースが SELECT プロジェクションを通じて enabled_columns を尊重するかどうか。
supports_row_filters(デフォルトは False)— ソースが保存された row_filters を適用するには True である必要があり、そうでない場合フィルターは無視されます。
has_managed_hogql_schema(デフォルトは False、Stripe、Paddle、Zendesk などの収益分析ソースでは True)— 固定のフィールドセットがマネージド HogQL スキーマを支え、これにより列選択が無効になります。これを無視する新しい課金/収益ソースは収益分析を壊す可能性があります。
cleanup_cdc_resources_on_deletion — CDC リソースをプロビジョニングするソース向けのベストエフォートのティアダウンフックです。ソースがそのようなリソースを作成する場合にのみオーバーライドしてください。

ベンダーの API バージョニング(supported_versions、default_version、api_docs_url、deprecated_versions、resolve_api_version())には、専用の warehouse-source-new-version スキルを使用してください — バージョン処理を自作しないでください。

必須のコーディング規約

@SourceRegistry.register で登録します。
再開可能な動作や webhook の動作が必要でない限り、SimpleSource[GeneratedConfig] を継承します。
API ソースは通常、エンドポイントリソースで table_format="delta" を返すようにします。
primary_keys はエンドポイント固有です(settings.py で宣言し、常に id とは限りません)。単一のフィールドが一意でない場合は複合キーを使用します。キーは親ごとではなく、テーブル全体で一意でなければなりません: ファンアウトする子エンドポイントはすべての親から行を集約するため、API がグローバルな一意性を明示的に文書化していない限り、キーに親識別子を含めてください(例: ["form_id", "token"])。一意でないキーは Delta テーブルに重複行を生み出し、その後のすべてのマージでそれらがマルチマッチします — マージは同期のたびに遅くなり、最終的にポッドが OOM になります。
可能な場合は新しいソースにパーティショニングを追加します:
- API ソース: 安定した datetime フィールドとともに partition_mode="datetime"。
- データベースソース: partition_count と partition_size。
変化しないパーティションキーを選んでください — created_at、dateCreated、firstSeen。updated_at や lastSeen は決して使用しないでください。
既知の永続的な失敗(401/403、無効/期限切れの認証情報、スコープの欠如)には get_non_retryable_errors() を追加します。
コメントは最小限にし、意図が明白でない場合にのみ記述します。
Python のインポートはモジュールの先頭に置き、関数内には置きません(循環インポートを解消するために必要な場合を除く)。

増分同期のガイダンス

supports_incremental=True を設定するのは、API がサーバー側のタイムスタンプフィルター（<field>_gte、since、modified_after など）を公開している場合のみにしてください。すべてのページを取得し、Python 側で既に見た行をスキップする「クライアント側カーソル」は増分ではありません — 実行のたびにすべてのページにアクセスするため、「増分」同期の API コストは結局フルリフレッシュと同一になります。API にサーバー側フィルターがない場合は、フルリフレッシュのみを提供してください。
API がサーバー側の時間フィルタリングをサポートしている場合は、それを使用し、db_incremental_field_last_value からマッピングしてください。
inputs.incremental_field を尊重してください — これはスキーマ設定でユーザーが選択したカーソルフィールドです。エンドポイントごとの INCREMENTAL_FIELDS は_提示される選択肢_のメニューです。INCREMENTAL_FIELDS[endpoint][0] に手を伸ばしてデフォルトを選び、ユーザーの選択を黙って上書きしないでください。
エンドポイントごとにソートの enum は異なります。 ?sorting=created_at（あるいはそれに類するもの）をグローバルにハードコードしないでください。各リストエンドポイントで許可されているソート値を API 仕様に対してかつ実際の API に対する curl スモークテストで検証してください — API はある選択肢のセットを文書化しておきながら別のものを黙って拒否したり、特定のリソースでは異なるタイムスタンプ列を使ったりすることが頻繁にあります。
ページネーション時には、安定した単調フィールドに対して ?sorting= を明示的に渡してください。 増分ソースでは、リクエストのソートは SourceResponse.sort_mode（通常は "asc"、API によって強制される場合のみ "desc" — stripe/stripe.py、github/settings.py を参照）と一致している必要があり、そうすることでパイプラインのカーソルウォーターマークが正しく進みます。フルリフレッシュソースでは、API の暗黙のデフォルトが不安定であったり、同期中に行が挿入されるにつれて変化したりする場合に、明示的なソートがページ境界でのスキップや重複を防ぎます。
API がカーソルページネーションのみをサポートしている場合でも、信頼できるのであれば増分フィールドを宣言し、マージのセマンティクスに重複排除を任せてください。
sort_mode は行が実際に到着する順序と一致していなければなりません — 仮定せず、検証してください。 パイプラインは sort_mode="asc" を信頼して、バッチごとに増分ウォーターマークをチェックポイントし、同期途中のワーカーシャットダウンを安全に許可します。API が新しい順で返すのに asc を宣言すると、ウォーターマークが破損し、再開のセマンティクスが壊れます。API の_デフォルト_のソートを確認してください（sort を渡せない場合に適用されます）。また、カーソルページネーションはソートパラメータを完全に拒否したり無視したりすることが多い点を覚えておいてください。
sort_mode="desc" は、エンドポイントが本当に昇順で返せない場合にのみ使用してください。降順ソースでは、新しい行より古い行を先に遡るために db_incremental_field_earliest_value を処理してください（Stripe を参照）。
増分ページネーションはウォーターマークで終了しなければなりません。 API によっては、時間ウィンドウフィルターとカーソルページネーションの併用を拒否するため、最初のページだけがウィンドウで絞られ、以降のページは履歴を無制限に遡ることになります。サーバーがすべてのページでフィルターを維持できない場合、ページャーは、ページ全体が db_incremental_field_last_value より古くなった時点でクライアント側で停止する必要があります（typeform/typeform.py:TypeformResponsesPaginator を参照） — そうしないと、増分同期のたびに各親の全履歴を再取得して再マージすることになり、これは API コストのバグであると同時に、同期ごとのメモリ増幅要因にもなります。
未知のエンドポイントはまずフルリフレッシュをデフォルトにしてください。増分を有効にするのは、安定したフィルターフィールドと API の順序付けセマンティクスを確認した後だけにしてください。
パーティションキーはエンドポイント名ではなく、レスポンススキーマに対して確認してください。

API 動作検証チェックリスト

エンドポイントのロジックを確定する前に、ドキュメントおよび実際の API に対して curl で検証してください（ドキュメントだけでは不十分です。API は未知のパラメータを黙って無視したり、古い enum を記載していたりすることが頻繁にあります）。

レスポンスの形状: リストか、オブジェクトか、ラップされたデータ（{"data": [...]}）か。
ページネーション: Link ヘッダーか、ボディのカーソルか、offset/page か。次ページの終了がどのように通知されるか。
順序の保証: 時間フィールドについて昇順/降順/未定義のどれか。また、何も指定しない場合の API の_デフォルト_の並び順。カーソル（before/after トークン）でページネーションする場合、API が sort や時間ウィンドウのパラメータを併用できるかを確認してください。多くの API はこれらを拒否または無視し、そのことが sort_mode と、増分同期におけるページネーションの終了方法の両方を決定づけます。
主キーの一意性のスコープ: id はグローバルに一意ですか、それとも親リソース内でのみ一意ですか。ファンアウトする子リソースについては、ドキュメントに別段の記載がない限り親ごとに一意とみなし、複合キーに親 id を含めてください。
エンドポイントごとのソート enum: 各リストエンドポイントはどの sorting= 値を受け付けますか。リソースごとに許可される enum が異なる API もあります。渡そうとしている値が 200 を返すことを curl で確認し、未来の日付のカットオフでプローブして、タイムスタンプフィルターが尊重されるのか黙って無視されるのかを確認してください。
サーバー側のタイムスタンプフィルター: <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 も尊重します。ステータスコードに対する 2 つ目の tenacity の @retry をフェッチにラップしないでください。それは複合します（例: トランスポートの 3 回の試行 × tenacity の 5 回の試行）。これは既存のソース全体で最も多くコピーされている間違いです。フレームワークまたはトラッキング対象のセッションを使用する場合、ステータスコードの再試行はすでに処理されています。何も書かないでください。
1 つのリクエストの再試行は 1 つの層が所有します。 RESTClient は自身のリクエストを再試行し（デフォルトで 5 回の試行）、それらのリクエストに対してアダプターの再試行をオフにします。トラッキング対象のセッションでの独自の呼び出しはアダプターの再試行を維持します（3 回の再試行、GET/HEAD/OPTIONS のみ）。
再試行の待機は上限があります。 1 つのリクエストは再試行に最大 DATA_WAREHOUSE_SOURCE_RETRY_BUDGET_SECONDS（600 秒）を費やします。RESTClient は、サーバーが DATA_WAREHOUSE_SOURCE_MAX_RETRY_AFTER_SECONDS（300 秒）を超える待機を要求した場合、そのリクエストを再試行可能なエラーで失敗させます。より長いレート制限ウィンドウを持つベンダーの場合は、クライアント設定で retry_after_max_seconds と retry_budget_seconds を設定してください。
バックオフやレート制限の保留のために time.sleep を呼び出さないでください。 common/interruptible_wait.py の interruptible_wait() を呼び出してください。ワーカーがシャットダウンを開始すると待機を終了し、実行を再開できます。safe_point= は、ステージングされたカーソルがソースの保持するどの行もカバーしていない位置からのみ渡してください。
トランスポートがカバーしない条件にのみ tenacity を追加してください。たとえば、再試行可能な HTTP ステータスではない、アプリレベルの「still processing」ボディなどです。追加する場合は、複合しないようにトランスポートの再試行を無効にしてください: make_tracked_session(retry=Retry(total=0))。
429 ではサーバー提供のレート制限リセットヘッダーを優先してください。トランスポートはすでに Retry-After を尊重します。カスタムの再試行は上限付きで決定的（stop_after_attempt）に保ち、明確な終端動作を持たせてください。
タイムアウトと再試行の設定は、調整しやすいようにモジュールの先頭付近に置いてください。
タイムアウトを指定しないリクエストには、default_request_timeout() のデフォルトの (connect, read) タイムアウト（30 秒、300 秒）が適用されます。エンドポイントに別の値が必要な場合は、クライアント設定で request_timeout を、またはセッション呼び出しで timeout= を渡してください。None は「デフォルトを使用する」を意味します。NO_REQUEST_TIMEOUT だけが、期限なしでリクエストを送信します。

上記のバックオフは、顧客が認証情報を所有している場合に適切な制御です。つまり、顧客自身のサードパーティアカウント上の自分の PAT / API キー / OAuth トークンであり、ほぼすべてのソースがこれに該当します。 PostHog は自分が所有していない予算を過剰に消費することはできないため、ソースで 429 / Retry-After を尊重すれば十分です。

例外は、PostHog が所有しプロセス間で共有する認証情報です。現在では、それが PostHog GitHub App のインストールトークンです（多くの PostHog サブシステムが、1 つのインストールごとの予算から同時に引き出します）。そこでは、事後対応的なバックオフでは不十分です。調整がなければ、同時実行の PostHog 呼び出し元が、429 が返る前に共有制限を集合的に超えてしまいます。そのような呼び出しは posthog/egress/ を経由する必要があります。これは Redis をバックエンドとする共有予算とテレメトリーであり、構造によってゲートされます。手書きの requests は決して使わないでください。そのような呼び出しを追加する前に /routing-outbound-api-calls を呼び出してください。 GitHub ソースが参照実装です。これはリミッターのキーを GitHub App のインストール ID（GitHub 自身の ID 空間における予算の所有者であり、PostHog の DB 行ではない）に設定し、顧客 PAT のパスはトークンに依存せずリミッターをスキップします。 api.github.com への生の呼び出しは github-api-calls-go-through-egress semgrep ルールによってブロックされるため、GitHub 形式のソースは構造によって egress パスに到達します。決定的な問いは決して「これはウェアハウスソースか?」ではなく、「誰がトークンを所有しており、同時実行の PostHog プロセスがそれを取り合って衝突する可能性があるか?」 です。

ファンアウトエンドポイント

ファンアウト = 親リソースを反復処理し、親ごとに子エンドポイントをクエリします。

単一ホップのファンアウトには依存リソースを優先してください。 親と、親フィールドに対して type: "resolve" を宣言する子を指定して rest_api_resources を使用します。共有インフラ（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 全体を再取得します。子エンドポイントは、その DependentEndpointConfig で parent_source="warehouse" を指定することでウェアハウスの親再利用をオプトインします。すると子は、親 API にアクセスする代わりに、親スキーマのすでに同期された Delta テーブルから親の行をストリーミングします（common/rest_source/warehouse_parent.py の iter_parent_pages_from_warehouse）。参照実装: Sentry の issue_tag_values カスタムイテレーターがリーダーを直接呼び出しており、増分ウォーターマークがスキャンを制限する場合のみウェアハウスモードになります。

要件と動作:

親は同じソースの選択可能なスキーマでなければならない — 親自身が Delta テーブルを生成する必要があります。
ソフト依存 — 子は親 API にフォールバックします。 親を宣言するには、ソースで get_required_parent_schemas をオーバーライドします（required_parents_from_endpoint_configs(ENDPOINTS, schema_name) に接続し、カスタムイテレータのエンドポイントには明示的なエントリを追加します）。このオーバーライドが唯一の宣言です。API を通じてこの関係が表面化することはないため、機能が検証されていないうちはスキーマペイロードにフィールドを追加しないでください。 API 側でも選択を制約するものは何もありません。親なしで子を有効化でき、子が同期している間に親を無効化または削除することもできます。import_data_activity_sync の _warehouse_parent_reuse_available が実行ごとに判断します — 親が存在しない、無効化されている、まだ初回同期が完了していない、または merge と full refresh 以外の同期タイプである場合、その実行は従来の親 API パスに送られるため、子をオプトインしても現在同期しているスキーマが壊れることは決してありません。同期途中の親はフォールバックを強制しません。resolve_parent_table_ref が Delta タイムトラベルを介して親の最後に完了したスナップショットに読み取りを固定するためです。子を有効化した副作用として親を有効化しないでください。親の同期は顧客の課金対象行にカウントされます。
小さい親は API に留まります。 MIN_WAREHOUSE_PARENT_ROWS（1,000 行）未満の親は、倉庫を開く価値がありません。Delta 読み取りには数秒の固定コストがかかり、小さなリストをページングするより高くつき、そのコホートは変換時に遅くなることが測定されています。このゲートは親テーブルの row_count を読み取るため、設定なしで実行ごとに適用されます。フィーチャーフラグはもうありません。ロールバックはリバートです。
厳密にストリーミング — 親テーブルを決してマテリアライズしないでください。 リーダーは、parquet 読み取りにプッシュダウンされた列プロジェクションを使用して、一度に 1 つの射影バッチをスキャンします。to_table、グローバルソート、または seen-set による重複排除を追加しないでください — 親は任意に大きくなる可能性があり、パイプライン全体はフルデータセットのメモリを回避するために存在します。呼び出し元のセマンティクスが親の順序に依存する場合（API がソートされた行を返していた場合）は、ソートするのではなく、順序のないストリームに対する行ごとのフィルターに作り替えてください（Sentry の issue_tag_values カットオフ処理を参照）。
使用可能な同期タイプは許可リストであり、拒否リストではありません。 merge と full refresh のみがキーごとに 1 行を保持します。append は同期ごとに行を蓄積し、CDC は変更履歴を保持するため、どちらかをストリーミングすると子が重複ごとに 1 回ファンアウトされ、重複排除には無制限の状態が必要になります。新しい同期タイプは _parent_unusable_reason で意図的にオプトインする必要があります。
値は API の JSON 型ではなく、Delta の物理型を持ちます。 タイムスタンプは ISO 文字列ではなく datetime として返り、ネストされたオブジェクトは 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 パスの有効な行セットを再現しなければならず、再現できない場合は親は失格となります。 ベンダーの list エンドポイントは通常、サーバー側で返すものを制限しますが、インクリメンタルな親テーブルはこれまでに見たすべての行を蓄積します — 無制限のスキャンは API パスでは決して行わない親に対してファンアウトし、古いスナップショットでは子の行数、課金ボリューム、実行時間を倍増させます。親をオプトインする前に、親を次の 3 つのケースのいずれかに分類してください:
- 親 API が本当に完全なコレクションを返す → フィルター不要。理想的な候補です。
- 制限が把握可能で表現可能 → DependentEndpointConfig に parent_row_filter を設定します（ParentRowFilter(field=..., not_older_than=..., not_before=...)）。述語は parquet 読み取りにプッシュダウンされ、NULL 行を保持し、文字列またはタイムスタンプの物理列に適応し、フィルター不可能なテーブルは eager 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=[...]) を使用してテーブルを eager に解決します — これは ORM 読み取りを行うため、パイプラインのイテレータ実行スレッドはアドホックな DB 接続に適した場所ではありません。また、親の Delta バージョンを固定するため、ファンアウト中に再同期する親が子の下の行をずらすことはありません。URI を再導出するのではなく、返された ref をリーダーに渡してください。そのようなソースは共有ファンアウトのテレメトリもスキップするため、親スイープが終了したときに log_fanout_parent_rows_consumed（sources/common/fanout_telemetry.py）を呼び出し、実行が実際に読み取った parent_source を指定してください。この行がなければ、倉庫ファンアウトのサイズを API パスのそれと比較できません。これは、リストが依然として返す行を親が欠落している場合の唯一の検出器です。
その resolve 呼び出しの周りで WarehouseParentTableNotFoundError をキャッチし、API パスを取ってください。 スキーマ行は、テーブルが読み取り不能（パージ、名前変更、またはファンアウト列の欠落）であっても完了した同期を主張する可能性があり、リーダーはジェネレータであるため、遅延して検証されたものはフォールバックが残っていないパイプラインの奥深くで発生します。これが、required_columns が解決中に eager に検証される理由です。フォールバックするときは、倉庫スナップショットにのみ意味のある動作（子の古い親に対する 404 → ignore、および倉庫スキャンがスキップする再開チェックポイント）もオフにして、実行が機能オフのパスと正確に一致するようにしてください。

OAuth の設定

OAuth を実装する前に、その連携がすでに存在するかどうかを確認してください — 新しいと結論づける前に、posthog/models/integration/ パッケージでサービス名を緩く検索してください。各種別は model.py にあり、OAuth の配線は oauth.py にあります。プロバイダーが独自のモジュールを持つのは、Slack、GitHub、Stripe のように OAuth 設定以外のビジネスロジックを伴う場合だけです。 __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", "")

連携の種別。
- posthog/models/integration/model.py の IntegrationKind 列挙型に追加します。
- posthog/models/integration/oauth.py の OauthIntegration.supported_kinds に追加します。
- 同じく oauth.py の oauth_config_for_kind() に elif kind == "your-source": return OauthConfig(...) ブランチを追加します。環境変数が空のときは NotImplementedError("<Source> app not configured") を発生させてください — これがフェイルクローズドのメッセージなので、シークレットの値が存在する前にコードと charts を出荷できます。
- プロバイダーのトークン応答にアカウント識別子がない場合(例: Resend)、アクセストークンの JWT をデコードし、クレーム(sub)から id_path / name_path を設定します。 integration_from_oauth_response の reddit/bing ブランチを参考にしてください。
クライアントの登録 + 認証情報のデプロイ。 プロバイダーへの OAuth クライアントの登録、リダイレクト URI(US/EU/dev/localhost)、charts の PR(環境変数を Web アプリ用の posthog-django-shared-secrets とワーカーの secret_env_app_specific ストアの両方に配線)、そして PostHog/secrets UI または CLI 経由で AWS Secrets Manager に値を書き込むこと — さらにこれらのうちどれをエージェントが自動化でき、どれは自動化してはならないか — はすべて 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`

ソース作成時に schema_name=None で呼ばれ(トークンが正当であることを確認するための 1 回の軽量なプローブ)、スキーマごとの incremental_fields アクションからは schema_name=<name> で呼ばれます(その特定のエンドポイントのスコープを確認)。

API が 401(不正なトークン)と 403(有効なトークン、スコープ不足)を区別する場合、ソース作成時には 403 を許容してください — ユーザーは同期したいエンドポイントのスコープだけを正当に付与することがあります。403 を再送出するのは schema_name が設定されているときだけです。同期時の 403 は get_non_retryable_errors() によって別途処理されます。

スキーマピッカーでテーブルごとのスコープ状態を表示するには、get_endpoint_permissions(config, team_id, endpoints) -> {name: None | reason} をオーバーライドします。各エンドポイントをプローブし、到達可能なら None を、そうでなければ短い理由を返します。database_schema アクションはこれを各テーブルの permission_error として表示するので、ユーザーはどのテーブルに追加のスコープが必要かを確認して選択を解除できます — これがソース作成を決してブロックしてはなりません。基底のデフォルトはすべて到達可能として報告します。

スコープ不足を表示するときは、それを名指ししてください — プロバイダーは通常それを示しています(Required access: `read_x` access scope.)。したがって、生の例外をそのまま出力したり、単なるテーブル一覧にまとめたりするのではなく、それを独自のメッセージに解析してください。テーブルごとのチェックがそのテーブルの同期に実際に何が必要かを反映するよう、同期クエリが必要とするフィールド(id だけでなく)をプローブしてください。プローブは狭く保ってください: 本当の拒否だけがスコープ不足です — スロットリング、5xx、ネットワークの一時的な問題はそうではないので、あらゆる例外を「権限不足」に分類するのではなく、それらを再試行可能なパスに通してください。

必要なトークンスコープを文書化する

API が OAuth スコープまたはリソースごとのアクセストークンを発行する場合、ソースが実際に呼び出すすべてのスコープを宣言して、ユーザーが何を付与すべきか分かるようにしてください。防御的にすべてのスコープを付与させるのは避けてください。

OAuth ソース: SourceFieldOauthConfig に requiredScopes を設定します（スペース区切りの文字列で、OAuth の scope パラメータの形式に一致します）。フロントエンドはこれをインテグレーションに付与されたスコープと差分比較し、不足がある場合は再接続アクションとともにユーザーに警告します。
非 OAuth ソース（PAT、API キー）: 検査すべきインテグレーションオブジェクトが存在しないため、代わりに caption にスコープを列挙します。キャプションは LemonMarkdown を通してレンダリングされるため、バッククォート、太字、リンクが機能します。

接続ホストフィールド（認証情報のリターゲティング）

ソースがシークレット（API トークン、パスワード）を保存し、それをユーザーが 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 を参照）。

ミックスイン

products/warehouse_sources/backend/temporal/data_imports/sources/common/mixins.py より:

SSHTunnelMixin — with_ssh_tunnel() コンテキストと、遅延トンネルオープン用の make_ssh_tunnel_func()。どちらも開くたびに SSH ホストとデータベースホストを再チェックします。
OAuthMixin — DB から Integration を取得する get_oauth_integration()。
ValidateDatabaseHostMixin — 内部 VPC IP をブロックする is_database_host_valid()（SSH トンネルが使用されている場合を除く）。

アイコン

PNG がリポジトリの慣例です（source.template はデフォルトで .png。約 800 個の png に対し約 58 個の svg）。SVG も受け入れられますが、標準ではありません。ファイルサイズは妥当な範囲に保ってください。
frontend/public/services/ に配置し、iconPath で /static/services/{name}.png（または .svg）として参照します。コミットする拡張子と一致させてください。
ソースのロゴがまだプロジェクトにない場合は、Logo.dev 経由で取得します。API キーはユーザーに尋ねてください — ハードコードしないでください。ユーザーが提供していない場合は、プレースホルダーをコミットするのではなく、ブロッカーとして報告してください。

テストの期待事項

ソーステストを書く前に references/testing.md を読んでください。 ソーステストはモノレポのテストケースの約5分の1を占めていました。 #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 のページネーター、認証、リトライ。
重複するカバレッジ: マルチページの走査の隣にある first-page または single-page のテスト、追加のレジューム状態のバリアント、および同じコードパスを共有するステータスやモードごとの個別テスト。代わりにパラメータ化してください。

tests/test_<source>.py に書いてください: 終端ページまでのマルチページの走査、フルリフレッシュに対するリクエスト形状のインクリメンタル、ソースが自身で行をソートまたはウィンドウ処理するときの yield の順序、トランスポートを介した実際のベンダーレスポンスに基づくエラーマッピング、保存された状態からのレジューム、およびエッジケース入力での行の形状。 tests/test_<source>_source.py は、ソースクラス自身が取る分岐にのみ追加してください。一般的な REST ソースには、10 から 25 のパラメータ化されたケースを含む 1 つのトランスポートモジュールが必要です。

実装チェックリスト

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 バケットを設定してください。
生成された config クラスがまだ空のまま: フィールドを更新した後に generate:source-configs を忘れている。
増分同期が誤動作する: フィールド名/型が間違っている、またはソートの想定が間違っている。
ビジーなテーブルで Pod が OOM になる: 主キーが実際には一意でない(通常はファンアウトの子に親の id がキーに含まれていない) — 重複行が蓄積し、マージのたびにそれらが複数マッチします。多くの場合、時間フィルターが 1 ページ目にしか適用されないため、同期のたびに全履歴を再走査する paginator と組み合わさります。
最新順で返す API に sort_mode="asc" を宣言している: ウォーターマークが最初のバッチの後でほぼ現在時刻にチェックポイントされ、同期途中のシャットダウンでデータの順序保証が失われます。
不正な認証情報で無限にリトライする: get_non_retryable_errors がない。
有効なトークンなのにソースが接続しない: validate_credentials(schema_name=None) がトークンだけでなくすべてのリソースのスコープをプローブするため、1 つの欠けたスコープ(多くの場合、ユーザーが同期しないテーブル上のもの)がソース全体をブロックします。作成時にはトークンのみをプローブし、テーブルごとのスコープは get_endpoint_permissions で報告してください。
再開可能な状態が保存されない: save_state の呼び出しを忘れている、またはパイプラインがまだ書き込んでいない行をカバーするカーソルで commit() を呼んでいるため、再開時にそれらがスキップされます。
カバーする yield の後に再開可能な状態を保存している: ハンドオフで最後のバッチの状態が失われるため、次の試行でそのバッチを再度読み取ります。posthog/test/repo_invariants/test_resume_state_staged_before_yield.py は、これをループ内で行う新しいジェネレーターで失敗します。そのベースラインは既存のものを列挙しており、短くなることしかありません。
デプロイが再開可能なソースで何時間も待機する: 行のないレスポンスをページングし続け、manager.safe_point() を決して呼び出さないため、ワーカーのシャットダウンを認識しません。
Webhook の行が反映されない: スキーマの is_webhook=False、または initial_sync_complete=False。
依存リソースのパスで KeyError: 静的パスのプレースホルダーを事前にフォーマットしてください(ファンアウトを参照)。
サイレントな切り捨てのリスク: ログ/メトリクスなしでページ上限に達する。
リファクタリングによるドリフト: エンドポイントの挙動変更後に、未使用の関数パラメーター/ヘルパーが残される。
エンドポイント config の dict における型のドリフト: 静的チェックを正確に保つために、ソースの型エイリアス(Endpoint、ClientConfig、IncrementalConfig)を使用してください。
パーティションキーの不安定さ: created_at ではなく updated_at を選んでしまい、同期のたびにパーティションが書き換わる。
Logo.dev のキーをハードコードしてコミットしてしまう: 実行時に必ずユーザーにキーを尋ねてください。

Markdown 対応

ログインしてコメント

基本情報

作者: posthog
リポジトリ: posthog/posthog
カテゴリ: データ・データベース
インストール数: 186
GitHub スター: 40,496

問題を報告