# Implementing Warehouse Sources

> Implement and extend PostHog Data warehouse import sources. Use when adding a new source under products/warehouse_sources/backend/temporal/data_imports/sources, adding datasets/endpoints to an existing source, or adding incremental sync, resumable imports, webhook ingestion, pagination, credentials validation, and source tests.

- Skill: `gabrielmoreira/implementing-warehouse-sources` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add gabrielmoreira/implementing-warehouse-sources`
- Raw SKILL.md: https://api.skillmd.com/api/skills/gabrielmoreira/implementing-warehouse-sources/raw
- Safety review: pending (external: skill-scanner PASS, skillspector CAUTION)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Data & Analytics
- Author: gabrielmoreira (https://skillmd.com/u/gabrielmoreira)
- Updated: 2026-09-09
- Page: https://skillmd.com/skills/gabrielmoreira/implementing-warehouse-sources

---


# Implementing Data warehouse sources

Use this skill when building or updating Data warehouse sources in `products/warehouse_sources/backend/temporal/data_imports/sources/`.

## Read first

Before coding, read:

- `products/warehouse_sources/backend/temporal/data_imports/sources/source.template` (the top-of-file TODOs are the bootstrap checklist; still verify target files against current source implementations, since the template can drift)
- `products/warehouse_sources/backend/temporal/data_imports/sources/README.md`
- `products/warehouse_sources/backend/temporal/data_imports/sources/SOURCES.md` — inventory of every registered source with its communication method (HTTP / vendor SDK / gRPC / DB protocol / webhook) and tracked-transport state. Skim this first to see how similar sources are wired and what state today's source you're touching is in. **Keep it in sync** — see "Updating SOURCES.md" below.
- `products/warehouse_sources/backend/temporal/data_imports/sources/common/base.py` — base classes (`SimpleSource`, `ResumableSource`, `WebhookSource`) and the `FieldType` union
- `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/` — the canonical reference for a new REST source.** It uses the shared `rest_source` framework (declarative `RESTAPIConfig` + `rest_api_resource`, framework auth + paginators, tracked+retrying transport) and is resumable — proof the framework covers the dominant "paginate a list endpoint and yield, resumably" shape. Read it first, alongside "Prefer the shared REST framework" below. Read `klaviyo/` or `github/` only as a _bespoke-transport_ fallback: they hand-roll their client for edge cases (custom query-string encoding, multi-level fan-out, JSON:API reshaping) that most sources don't have — don't copy that boilerplate into a source that doesn't need it. For dependent-resource fan-out (parent→child with `type: "resolve"`), also read `products/warehouse_sources/backend/temporal/data_imports/sources/common/rest_source/__init__.py` and `config_setup.py` (e.g. `process_parent_data_item`, `make_parent_key_name`).
- For webhook-capable sources, read `products/warehouse_sources/backend/temporal/data_imports/sources/stripe/source.py` as the reference implementation.

## Picking the right base class

Every new source **must** inherit from one (or a combination) of these:

- **`SimpleSource[Config]`** — default for straightforward pull-based APIs where each run fully iterates the endpoint.
- **`ResumableSource[Config, ResumableData]`** — **preferred for any new API-backed source whose underlying API supports resumption** (cursor/link-header pagination, time windows, offset tokens, or any other deterministic way to pick back up where we left off). If the API gives us a next-page token, a `Link` header, or a stable time filter, use `ResumableSource`. This lets Temporal resume after heartbeat timeouts without restarting from scratch. The manager persists state to Redis (24h TTL).
- **`WebhookSource[Config]`** — only when the source can push events to us (e.g. Stripe webhook endpoints). Typically combined with `ResumableSource` so the initial backfill is resumable and subsequent deltas come via webhook.

Combine by multiple inheritance when both apply, e.g.:

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

Rule of thumb:

- Pull-only API, no cursor we can persist → `SimpleSource`.
- Pull-only API with any cursor/next-page/time-filter we can save between runs → `ResumableSource`.
- Source can call us back with change events → add `WebhookSource` on top of whichever pull base fits.

Databases and file-transfer sources (SFTP, S3) stay on `SimpleSource` unless there's a clear reason otherwise.

## Prefer the shared REST framework

Most REST sources should be built on the shared `rest_source` framework
(`common/rest_source/`), not a hand-rolled client. It already provides — so you write **none** of it:

- **Tracked, retrying transport** — `RESTClient` defaults to `make_tracked_session()` and retries
  `429` + transient `5xx` honoring `Retry-After`. No `tenacity`, no `RetryableError`, no fetch loop.
- **Paginators** (`rest_source/paginators.py`, chosen by string/dict in the config, not hand-written):
  `single_page`, `header_link`, `json_response` (next-URL in body), `cursor`, `offset`, `page_number`.
- **Auth** (`rest_source/auth.py`): `bearer`, `api_key` (header/query/cookie), `http_basic`, `oauth2`
  (customer-owned client-credentials/refresh). Each redacts its own secrets — no `_get_headers` builder.
- **Incremental params, `data_selector`, response actions, resume** (`resume_hook` /
  `initial_paginator_state`), and **parent/child fan-out** (`fanout.build_dependent_resource`).

`chargebee/` is the canonical example (declarative endpoints + framework auth + resume). `zendesk/`
shows multi-endpoint + `data_selector`; `attio/` shows cursor pagination.

**When hand-rolling is justified** (read `klaviyo/` then): the API needs query strings the framework
can't produce (literal brackets/operators, e.g. `filter=greater-than(...)`, `page[size]`);
multi-level (2+ deep) fan-out; or per-item reshaping the `data_selector` can't express (e.g.
flattening JSON:API `attributes` into the row root). Single-level fan-out and per-item maps are
supported declaratively — don't hand-roll for those. If you must hand-roll, still ride
`make_tracked_session()` and do **not** add a second status-code retry layer (see "Retry and throttling").

## End-to-end workflow for a new API source

Follow this order. Each step maps to TODOs in `source.template`.

1. **Survey the source.** Pick the endpoints a user will actually want. Cross-reference:
   - Airbyte: <https://airbyte.com/connectors> (connector pages often link to source code — useful reference)
   - Fivetran: <https://www.fivetran.com/connectors>
   - Stitch: <https://www.stitchdata.com/docs/integrations/>
     Find the official API docs or OpenAPI spec, and **work out the vendor's latest generally-available API version before you write any request code** — that is the version the source must be built against. Check the vendor's changelog, versioning, or deprecation page, not just whichever page ranked first; docs sites routinely default to an older version, and Airbyte/Fivetran connectors are often years behind. See "Vendor API version metadata" for what counts as latest and what to do when the newest channel isn't GA.
2. **Bootstrap the source.** Copy the template and wire up the enum/type references:

   ```sh
   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
   ```

   Then update the two hand-edited files (the template still lists `posthog/schema.py` too, but that file is regenerated by `pnpm run schema:build` in step 12 — don't maintain it by hand):
   - `ExternalDataSourceType` at `products/warehouse_sources/backend/types.py` — follow the existing convention in that file: `ALL_CAPS` with **no underscores** between words (e.g. `ACTIVECAMPAIGN`, `APPLESEARCHADS`), value is `PascalCase`
   - `externalDataSources` at `frontend/src/queries/schema/schema-general.ts` — **PascalCase, identical to the `ExternalDataSourceType` value** (e.g. `'ActiveCampaign'`, `'GoogleAds'`, `'CustomerIO'`). NOT kebab-case. (The only kebab-case identifier in the flow is the optional `featureFlag="dwh-{source_name}"`.)

3. **Pick the base class** (see above) and rename the class / `source_type` return.
4. **Define `get_source_config`** — name, **category** (required — see "Source category & keywords"), label, caption, docsUrl, iconPath, fields, and optional `keywords`. Use appropriate field types (see below). Also set the vendor API version metadata class attributes — see "Vendor API version metadata".
5. **Register** the source — add an import line to `products/warehouse_sources/backend/temporal/data_imports/sources/__init__.py` and include it in `__all__`. (The `@SourceRegistry.register` decorator on the class handles runtime registration.)
6. **Run the config generator**: `pnpm run generate:source-configs`. Confirm the new config class appears in `products/warehouse_sources/backend/temporal/data_imports/sources/generated_configs/<your_source>.py` (one generated module per source; the package `__init__.py` is hand-written and never regenerated). **Do not edit generated modules by hand.** Every time you change `get_source_config.fields`, re-run the generator.
7. **Swap the generic `Config` type** in `source.py` for the generated `{Source}SourceConfig` class.
8. **Implement**: `validate_credentials`, `get_schemas`, `source_for_pipeline` (plus `get_resumable_source_manager` / `get_webhook_source_manager` as needed).
9. **Split transport logic.** Put API client, paginator, row normalization, and `SourceResponse` assembly in `{source}.py`. Keep endpoint catalog/incremental fields/primary keys/partition defaults in `settings.py`.
10. **Add icon.** Place at `frontend/public/services/{source}.png` — **PNG is the repo convention** (~800 png vs ~58 svg, and `source.template` defaults to `.png`). SVG is accepted but not the norm; set `iconPath` to match whichever extension you commit. If the logo isn't already committed, fetch from [Logo.dev](https://docs.logo.dev/introduction) — **ask the user for the Logo.dev API key**; do not hardcode one. Logo.dev's image API returns PNG (not SVG). Keep file size reasonable.
11. **Run migrations.** `DEBUG=1 python manage.py makemigrations && DEBUG=1 ./bin/migrate` (only needed if a new enum value triggers a Django migration).
12. **Rebuild schema types**: `pnpm run schema:build`. This updates `posthog/schema.py` from `schema-general.ts` and makes the source appear in frontend dropdowns. Re-run whenever `schema-general.ts` changes.
13. **Release status — a finished source has no `unreleasedSource` flag.** The default for the deliverable this skill produces is **no `unreleasedSource`** — a completed, working source ships visible and connectable. You don't need anyone's sign-off to ship it released; that's just the finished state. The scaffolded stub ships with `unreleasedSource=True` pre-set, so deleting that line is part of finishing the source — go ahead and remove it. (Why it matters: `unreleasedSource=True` **hides the connector from users entirely** — the frontend filters out every source where it's truthy; see `DataWarehouseQueryVariant.tsx`, `InlineSourceSetup.tsx`, and the "coming soon / Notify me" path in `nonHogFunctionTemplatesLogic.tsx`.)

    **Deleting that line is mandatory, and it is not gated on anything you can't do in your environment.** In particular, "I couldn't curl the live API" or "I couldn't verify against a real account" is NOT a reason to keep the flag — that is exactly what `releaseStatus=ReleaseStatus.ALPHA` is for (a soft "new, lightly tested" label on a _visible_ source). The only time `unreleasedSource=True` legitimately stays is when the source physically cannot sync yet because it is being landed across several PRs and the implementing code isn't all there. A source with working `get_schemas` / `source_for_pipeline` and passing tests is finished — the flag comes out. **Never write a test that asserts `unreleasedSource is True`** — that locks the bug in and is what kept 166 finished sources hidden until they had to be released in bulk.

    So a newly finished, tested source ships with:
    - **no `unreleasedSource`** (visible and connectable),
    - `releaseStatus=ReleaseStatus.ALPHA` for a new source that hasn't been extensively tested (`ReleaseStatus.BETA` once rough edges are ironed out; `ReleaseStatus.GA`, or omit `releaseStatus` entirely, for general availability) — a soft label on a _visible_ source, not a gate,
    - optional `featureFlag="dwh-{source_name}"` (kebab-case) **only** if you want a controlled rollout to flagged users instead of releasing to everyone.

    Whenever you set `releaseStatus`, use the `ReleaseStatus` enum from `posthog.schema` — never a bare string literal. Add `ReleaseStatus` to your existing `from posthog.schema import (...)` block.

14. **Document the source.** Write or update the user-facing doc on posthog.com following the
    `/documenting-warehouse-sources` skill (template, shared snippets, `<SourceParameters />` +
    `<SourceTables />`). Ensure `docsUrl` in `get_source_config` matches the doc filename
    (kebab-case), and — if `get_schemas` is a static endpoint catalog — set
    `lists_tables_without_credentials = True` (see below) so the doc's Supported tables section
    renders. A finished source ships with a consistent doc, not a stub.
15. **Delete the template TODO comments** before PR.

## Source architecture contract

For API-backed sources, use this split:

- `source.py`: source registration, source form fields, schema list, credential validation, resumable/webhook manager wiring, pipeline handoff.
- `settings.py`: endpoint catalog, incremental fields, primary key, partition defaults.
- `{source}.py`: API client/auth, paginator, request params, row normalization, and `SourceResponse`.

This keeps endpoint behavior declarative and easy to extend.

### Source behaviour goes in the source, never in the API layer

The `warehouse_sources` presentation layer (`products/warehouse_sources/backend/presentation/views/external_data_source.py`, `external_data_schema.py`) must stay source-agnostic.
Do **not** add `if source_type == ExternalDataSourceType.X` / `source.is_direct_<engine>` branches there — a CI guard (`.github/scripts/check-dwh-source-agnostic.py`) blocks new ones.

When a source needs behaviour the API must invoke, expose it on the source instead:

- **A boolean/value the API reads** → add a flag on `_BaseSource` with a safe default (like `supports_column_selection`, `connection_host_fields`, `has_managed_hogql_schema`), and let the API branch on the flag.
- **Methods only some sources have** (CDC, xmin, webhooks, custom manifests) → a capability mixin the source opts into; the API dispatches with `isinstance(source, <Capability>)`.
- **Direct-query engine behaviour** (how a SQL engine resolves a table location, builds its `DataWarehouseTable`, maps columns) is keyed on the engine, not the source type — dispatch on `source.direct_engine` through the engine adapter/registry (`posthog/hogql/direct_sql/` for query concerns, the `data_warehouse` engine registry for materialization), never `source_type`.

Keep source-domain semantics (how to talk to the engine, how it names things, whether filters push down) on the source; the warehouse-domain work it drives (`DataWarehouseTable` rows, managed viewsets, hog functions) stays in `data_warehouse`, keyed off what the source or adapter returns.
Source capabilities never import `data_warehouse` types.
See `products/data_warehouse/backend/presentation/README.md`.

For REST sources that mix top-level and fan-out endpoints, keep endpoint metadata in `settings.py` and route in `{source}.py` with this priority:

1. endpoint-specific custom iterators (only when required),
2. generic fan-out helper path,
3. top-level endpoint path.

## Canonical descriptions (semantic enrichment)

After a table syncs, a background activity (`workflow_activities/enrich_table_semantics.py`) writes
`WarehouseColumnAnnotation` rows describing each table/column, surfaced to the AI agent. For
fixed-schema sources (SaaS APIs) the schema is the same for everyone, so document it **once** from the
official API docs instead of paying an LLM to re-derive it per team. These curated descriptions are
authoritative — they're applied directly (`description_source="canonical"`) and never sent to the LLM.

Add a `canonical_descriptions.py` **accompanying the source** (sibling of `source.py` / `settings.py`):

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

Then override the hook on the source class with a lazy import of the sibling file:

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

Rules:

- Key entries by the **endpoint/schema name** `get_schemas` returns (matches `ENDPOINTS`), not the
  prefixed warehouse table name.
- Source descriptions from the **official API docs**, not guesses. Partial coverage is fine — any
  missing endpoint, column, or table-level `description` falls back to the LLM, which is given the
  source name, endpoint, `docs_url`, and column data types.
- Optional and only meaningful for fixed-schema sources. SQL sources (arbitrary user schemas) ship
  nothing — the base hook returns `{}`.
- Don't touch `source.py`/`settings.py` transport logic — this is purely additive metadata.

## Publishing the table catalog to public docs

The posthog.com docs render a **Supported tables** section via a `<SourceTables />` component fed by the
`public_source_configs` API, which calls `get_documented_tables()` on each source. The base
implementation lists tables from `get_schemas` (merged with `canonical_descriptions`) **only when the
source opts in**:

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

Set this to `True` **only** when `get_schemas` iterates a static endpoint catalog with **no I/O** — no
network, no DB, no credentials (the common fixed-schema SaaS pattern: `for endpoint in ENDPOINTS`). The
endpoint builds a placeholder config and calls `get_schemas` with no real credentials, so a source that
connects to discover schemas (SQL, file storage, MongoDB, ad platforms that list accounts) must leave
this `False` (the default) — otherwise it would try to connect to an empty host, hang, or close the DB
session. When `False`, the docs render a generic "discovered from your account" note instead.

The richer the table list, the better the docs — so pair this with `canonical_descriptions.py`
(table/column descriptions). Verify the rendered output via the API:
`GET /api/public_source_configs` → your source → `tables`.

## Source category & keywords

Every source **must** set `category` on its `SourceConfig` — it groups the source in the new-source wizard
catalog (a category rail + tile grid). A test (`tests/test_source_categories.py`) fails if any registered
source has no category, so this is non-optional. Import the enum from `posthog.schema`:

```python
from posthog.schema import DataWarehouseSourceCategory
...
return SourceConfig(
    name=SchemaExternalDataSourceType.STRIPE,
    category=DataWarehouseSourceCategory.PAYMENTS___BILLING,
    keywords=["billing", "subscriptions"],
    ...
)
```

Pick the single closest bucket. The enum members (note the triple underscore where the label has " & "):

- `DATABASES` — OLTP/OLAP databases, warehouses, data streams (Postgres, Snowflake, BigQuery, Kafka, …)
- `FILE_STORAGE` — object/file stores & file transfer (S3, Azure Blob, GCS, Google Drive, SFTP, …)
- `ADVERTISING` — ad platforms & mobile attribution (Google Ads, Meta Ads, Reddit Ads, Adjust, …)
- `MARKETING___EMAIL` — email/SMS/marketing automation (Klaviyo, Mailchimp, Braze, SendGrid, …)
- `CRM` — CRM & sales intelligence (HubSpot, Salesforce, Attio, Pipedrive, ZoomInfo, …)
- `SALES` — sales engagement/enablement, contracts (Salesloft, Outreach, Gong, DocuSign, …)
- `CUSTOMER_SUPPORT` — helpdesk/support/CX (Zendesk, Intercom, Freshdesk, Front, …)
- `PAYMENTS___BILLING` — payment processors & subscription billing (Stripe, Chargebee, PayPal, …)
- `FINANCE___ACCOUNTING` — accounting/ERP/expense/spend (QuickBooks, Xero, NetSuite, SAP ERP, …)
- `ANALYTICS` — product/web/marketing analytics & experimentation (Amplitude, Mixpanel, GA, …)
- `ENGINEERING___MONITORING` — dev tooling, CI, error/uptime monitoring, feature flags, identity/auth (GitHub, Datadog, Sentry, LaunchDarkly, Auth0, …)
- `PRODUCTIVITY` — project mgmt, docs, forms, scheduling (Notion, Airtable, Jira, Linear, Typeform, …)
- `HR___RECRUITING` — HRIS/ATS/payroll/people (Ashby, Greenhouse, BambooHR, Workday, Gusto, …)
- `COMMUNICATION` — messaging/meetings/telephony/social (Slack, Zoom, Microsoft Teams, Twilio, …)
- `E_COMMERCE` — online store/commerce (Shopify, WooCommerce, BigCommerce, …)

The category list is the source of truth in `frontend/src/queries/schema/schema-general.ts`
(`dataWarehouseSourceCategories`); `pnpm run schema:build` regenerates the Python `DataWarehouseSourceCategory`
enum. Adding a **new** category means editing that array and rebuilding — don't invent ad-hoc strings.

`keywords` is an optional list of lowercase search aliases — only add when the source has a common acronym or
alternate spelling a user might type (e.g. `["ga4", "ga"]`, `["sql server"]`, `["facebook ads"]`). Skip it when
the name already obviously matches; don't add noise.

## Self-driving Inbox candidacy (issues / tickets / conversations)

Some sources are also candidates for the **Self-driving Inbox** — the feature that watches a synced
table of _actionable records_ and emits findings into the PostHog Desktop Inbox. Shipped today: GitHub,
Linear, Zendesk, pganalyze, and Jira.

The signal is the **table you sync**, not the vendor: a source is an inbox candidate when one of its
tables is a stream of records a human (or agent) triages one by one — an `issues`, `tickets`, or
`conversations` table. These live under the support/helpdesk (`CUSTOMER_SUPPORT`), issue-tracker and
monitoring (`ENGINEERING___MONITORING`), and some project-tool (`PRODUCTIVITY`) categories. Analytics,
billing, ad-platform, CRM, and raw database sources are **not** inbox candidates — they sync facts to
query, not a work queue to act on. If the source you're building has no such table, there's nothing to
do here.

Wiring a source into the inbox is a **separate, additive piece of work** with its own skill —
`/adding-inbox-sources` — and it changes nothing in this skill's deliverable. It only becomes possible
once the data-warehouse source exists (which is exactly what this skill produces), so build and ship the
source first. That skill touches three surfaces: a server-side "signals scout" emitter plus a registry
entry and `SignalSourceProduct` enum in this repo (`products/signals/backend/`), the inbox UI in the
separate `posthog/code` repo, and the `npx @posthog/wizard self-driving` onboarding flow in
`PostHog/context-mill`. Read `/adding-inbox-sources` before starting — none of that plumbing belongs in
the source's own `products/warehouse_sources/` code.

## Vendor API version metadata

Every source declares three class attributes (on the source class body, alongside `lists_tables_without_credentials`)
describing the vendor's API version.
The framework (`common/base.py`) records the version each `ExternalDataSource` runs against so old pins keep working
and deprecations can be surfaced;
`sources/tests/test_source_versions.py` enforces the invariants below across every registered source, so a new
source that gets these wrong fails CI.

Two cases:

- **The vendor exposes a real, pinnable API version** — a URL path segment (`/v3/`, `/2/`), a required version
  header value (a dated `2022-11-28`), a dated query/version param, or a named release. Declare all three:

  ```python
  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)
  ```

  **Build the source against the vendor's latest generally-available version, and pin that.** A new source starts
  on one version and every customer who connects it lands there, so shipping on an older version means shipping a
  migration someone has to run later. Two rules, and they must agree:

  1. Write the request code against the newest GA version the vendor offers.
  2. Declare **the version that code actually calls** (the base URL path, a version header, or a version constant
     in `settings.py` / `{source}.py`). Never declare a version the code doesn't send — that pin is a lie the
     framework can't detect, and it makes the deprecation warnings and the upgrade path wrong for every customer.

  If you can't reach the newest version — it's preview/beta/unstable/RC, it's gated behind an application or a
  paid tier, or its response shapes aren't implemented yet — build against the newest GA version you can actually
  call, pin that, and say why in a comment on the class. "Latest" means latest stable: don't pin Shopify's
  `unstable`, a vendor's `-rc` channel, or a version whose docs are still marked preview.

  Examples already in the tree: Anthropic `("2023-06-01",)` (dated `anthropic-version` header),
  ActiveCampaign `("v3",)` (`/api/3` path segment), Alguna `("2026-04-01",)` (dated version header).
  A source that later gains a second version declares them oldest→newest — GitHub `("2022-11-28", "2026-03-10")`,
  HubSpot `("v3", "2026-03")` — but that's the `/warehouse-source-new-version` skill's job, not this one.

- **The vendor has no meaningful API versioning** — set only `api_docs_url`; leave `supported_versions` /
  `default_version` at the framework default (`("v1",)`, the `UNVERSIONED_API_VERSION` sentinel). A bare `/v1/`
  that has never changed and isn't a documented version choice is this case.

Rules:

- `default_version` must equal the single entry in `supported_versions`, and `api_docs_url` must be `https://`.
- Use the vendor's exact version string; never invent one.
- **Never ship a new source on a version the vendor has already deprecated or given a sunset date.** A brand-new
  source with a `deprecated_versions` entry covering its only version is a bug — it means the source was written
  against the wrong version. `test_source_versions.py` fails the build if `default_version` is deprecated.
- Prefer an `api_docs_url` that points at the vendor's versioning/changelog page over a generic API landing page —
  it's where the next version gets announced, and it's what the next person checks before repinning.
- Don't hardcode a fallback version in the transport/request layer — resolve it from the source class
  (`self.resolve_api_version(inputs.api_version)`), which already falls back to `default_version`.
- Adding support for a **new** vendor version later, or **deprecating** an old one, is the
  `/warehouse-source-new-version` skill — not this one.

## Source fields (the form the user fills in)

Defined in `get_source_config.fields`. All field types live in `posthog/schema.py` and are unioned as `FieldType` in `products/warehouse_sources/backend/temporal/data_imports/sources/common/base.py`.

- `SourceFieldInputConfig` — basic input (`text`, `email`, `number`, `password`, `textarea`). Rendered as `<LemonInput />`.
- `SourceFieldSwitchGroupConfig` — toggle that reveals a sub-group of fields. Use for optional feature blocks.
- `SourceFieldSelectConfig` — dropdown. Options can carry sub-`fields` shown when selected (use for alternative auth methods — e.g. API key vs OAuth).
- `SourceFieldOauthConfig` — OAuth via `Integration` model. See OAuth section.
- `SourceFieldFileUploadConfig` — file upload (JSON). Use `keys=["..."]` allow-list or `"*"`.
- `SourceFieldSSHTunnelConfig` — renders SSH tunnel sub-fields; adds `ssh_tunnel: SSHTunnel` to the config with helpers.

Guidelines:

- Multiple auth methods → `SourceFieldSelectConfig` with child `fields` per option.
- Optional toggles → `SourceFieldSwitchGroupConfig`.
- Confidential fields must use `SourceFieldInputConfigType.PASSWORD`. The serializer derives sensitive vs nonsensitive keys automatically from the field definitions — you do not need to maintain an allow-list elsewhere.

## Implementing `source_for_pipeline`

Return a `SourceResponse` directly. **Do not** use `dlt_source_to_source_response` for new sources — DLT is being removed.

Prefer yielding data in the shape the API returns it. No custom dataclasses, no heavy parsing. Yield either `dict`, `list[dict]` (preferred when possible), or a `pyarrow.Table`. The pipeline buffers and batches for you.

**Default to yielding raw `dict` / `list[dict]` and let the pipeline batch for you.** The pipeline already runs a `Batcher` (`pipelines/pipeline_v2/pipeline.py`) at 5000-row / 200 MiB thresholds, so the common case needs no batcher of its own. Reach for `pyarrow.Table` only when you already have arrow-shaped data (e.g. a ClickHouse adapter). A source _may_ instantiate its own `Batcher` with **smaller** thresholds (e.g. `chunk_size=2000, chunk_size_bytes=100 * 1024 * 1024`, as klaviyo and ~70 other sources do) when it deliberately wants a tighter memory footprint for large/wide rows — that's a valid choice, not the default. What to avoid is a second _full-size_ batcher, which just double-buffers with no win.

For pyarrow tables, cap in-memory rows at ~200 MiB or ~5000 rows. Use helpers like `table_from_iterator()` / `table_from_py_list()` from `products/warehouse_sources/backend/temporal/data_imports/pipelines/core/arrow_utils.py`.

**URL construction:** use `urllib.parse.urlencode` for query strings. Don't use `requests.Request(...).prepare().url` — `PreparedRequest.url` is typed `Optional[str]` and the typical workaround (`prepared.url or f"..."`) carries an unreachable fallback. `urlencode` is shorter, dependency-free, and produces identical output for ASCII-safe params.

### Resumable source pattern

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

In the transport function:

```python
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)
    # yield batch
    next_url = data.get("links", {}).get("next")
    if not next_url:
        break
    manager.save_state(MyResumeConfig(next_url=next_url))
    url = next_url  # advance before the next fetch, otherwise we loop on the same page
```

Save state **after** yielding each batch, not before — so if we crash we re-yield the last batch (merge dedupes on primary key) rather than skipping it.

### Webhook source pattern

- Implement `webhook_template` returning a `HogFunctionTemplateDC` that transforms incoming webhook payloads.
- Implement `webhook_resource_map` mapping our schema name → external object type.
- Implement `create_webhook`, `delete_webhook`, `get_external_webhook_info` if the API allows programmatic webhook management. Otherwise return a failed result and provide a `webhookSetupCaption` explaining manual setup.
- Add `webhookFields` to `SourceConfig` for post-setup inputs (e.g. signing secret).
- In `source_for_pipeline`, call `self.get_webhook_source_manager(inputs)` and pass its iterator alongside the pull iterator so a single sync pulls historical + webhook-delivered rows.
- Populate `SourceSchema.supports_webhooks=True` only for endpoints where webhooks are actually viable (usually incremental/append-only ones).
- **De-dupe within a webhook batch with a `table_transformer`.** `WebhookSourceManager.get_items()` takes an optional `table_transformer: Callable[[pa.Table], pa.Table]` applied after the raw webhook payloads are deserialized into row dicts. Delta merge only de-dupes _across_ syncs (on `primary_keys`), not within a single source batch — so when one batch can carry multiple events for the same object (e.g. `customer.created` then `customer.updated`), pass a transformer that keeps only the latest version per id. Reference: `_webhook_table_transformer` in `stripe/stripe.py`, wired via `webhook_source_manager.get_items(table_transformer=_webhook_table_transformer)` in `stripe_source`. It groups rows by `object.id`, keeps the one with the greatest event `created` timestamp, and rebuilds the table shaped like the underlying object (ready to merge on `primary_keys=["id"]`).

## Multi-schema SQL database sources

SQL DB sources (Postgres, MSSQL, Snowflake, Redshift today) can import tables from **every namespace (schema) in one connection**: a blank namespace field discovers tables across all non-system namespaces, the wizard groups them by namespace, and sync writes one warehouse table per `namespace.table`. Reference implementation: `postgres/postgres.py` + `PostgresImplementation`; the shared seam lives in `common/sql/`.

The capability marker is the source's `schema` field being **optional** (`required=False`) in `get_source_config` — `is_multi_schema_capable_sql_source()` (`products/data_warehouse/backend/sql_warehouse_migration.py`) keys off it, so flipping the field optional is what turns on the viewset migration behavior. Treat `None` / `""` / whitespace as "all namespaces" (`normalize_namespace` in `common/sql/location.py`) and never emit `WHERE table_schema = ''`.

Checklist for bringing a SQL source to multi-schema parity:

1. **Namespace field optional** — `required=False` on the `schema` field, rerun `pnpm run generate:source-configs`. Keep `database` required: the database/catalog stays fixed per connection.
2. **Multi-namespace discovery** — in `get_columns`, `get_primary_keys`, index/row-count/foreign-key helpers: when the namespace is blank, drop the `WHERE table_schema = <ns>` predicate (excluding system namespaces like `information_schema`, `pg_catalog`, `sys`) and return **qualified display names** (`namespace.table`). Keep the single-namespace fast path when the field is set.
3. **Implement `get_source_metadata`** — return `SourceMetadata(catalog_by_table, schema_by_table, table_name_by_table)` keyed by the qualified display name. `SQLSource.get_schemas` stamps it onto each `SourceSchema`, and `reconcile_schema_metadata` persists it into `ExternalDataSchema.sync_type_config["schema_metadata"]`.
4. **Per-row routing in `build_pipeline`** — resolve `(schema, table_name, response_name)` with `resolve_source_location` (`common/sql/location.py`): per-row metadata → dotted-name self-heal → config namespace. Run SQL against the resolved schema + **unqualified** table; set `SourceResponse.name = response_name` (`dwh_storage_key or schema.name`, normalized) — never the bare table name, or the row's Delta path moves and orphans synced data.
5. **Thread the resolved namespace through every streaming/stats helper** — table metadata, row stats, average row size, partition settings, chunk size, primary-key lookup all take `(schema, table)`. Missing one degrades silently (no partitioning / wrong stats).
6. **Never feed a dotted display name to an identifier quoter** — `quote("a.b")` yields one wrong identifier. Split into `(schema, table)` first and use `quote_qualified` (`common/sql/identifiers.py`).
7. **Naming layers are derived, never stored** — display name `analytics.users`; S3/Delta subdir `analytics_users` (normalized `response_name`); HogQL table `{prefix}_analytics_users`. The one stored exception is `dwh_storage_key`, which pins a migrated legacy row to its original Delta path.
8. **Legacy migration is capability-driven, not source-type-gated** — when a user clears the namespace on an existing single-schema source, `sql_warehouse_migration.py` renames rows to qualified form and stamps `dwh_storage_key`, preserving synced data with no re-sync. Don't add `source_type == "..."` branches to the shared layer.
9. **Tests** — two namespaces with the same table name stay distinct end to end; blank-namespace discovery excludes system namespaces; per-row routing hits the right namespace; legacy single-namespace sources keep working; migrated rows keep their legacy Delta path.

Discovery cost: `validate_credentials` and `database_schema` run discovery with no name filter, so a blank namespace on a catalog with hundreds of schemas must not issue per-table queries per namespace — batch the listing queries or cap enumeration (see Snowflake's `SHOW PRIMARY KEYS` handling).

## Outbound HTTP must go through the tracked transport

Every HTTP call from `products/warehouse_sources/backend/temporal/data_imports/sources/**` must go through `make_tracked_session()` (from
`products.warehouse_sources.backend.temporal.data_imports.sources.common.http`). The tracked session attaches `team_id`, `source_type`,
`external_data_source_id`, `external_data_schema_id`, and `external_data_job_id` to every outbound request's
log line and OTel metric, and participates in opt-in sample capture.

- For raw `requests` usage: `make_tracked_session(headers=..., retry=...)` returns a `requests.Session`. Use
  `session.get/post/...` instead of the module-level `requests.get/...` shortcuts.
- **Redact secrets from the captured samples.** Pass `redact_values=(api_key, token, ...)` to
  `make_tracked_session(...)` so the tracked transport masks those literal values in logged URLs, headers,
  and sampled bodies — important for keys that ride in a query param or an odd header name. `rest_source`
  auth classes do this automatically (each implements `secret_values()`); only raw-session sources need to
  pass `redact_values` themselves.
- For sources that already go through `rest_source.RESTClient`: it defaults to a tracked session
  automatically; no change needed.
- For vendor SDKs that accept a session/HTTP-client hook (Stripe `RequestsClient(session=...)`,
  gspread `authorize(credentials, session=...)`, BigQuery via `AuthorizedSession` + `TrackedHTTPAdapter`),
  inject one. Reference patterns live in `stripe/stripe.py`, `google_sheets/google_sheets.py`, and
  `bigquery/bigquery.py`.
- For vendor SDKs with no injection seam (today: `bingads`, `linkedin-api`'s `RestliClient`), add a
  `# nosemgrep: data-imports-http-transport-...` pragma with a one-line reason and record the source as
  `⚠️ Vendor SDK` in `SOURCES.md`.
- gRPC SDKs are **not** exempt — they have their own tracked transport (see below).

CI enforces this via `.semgrep/rules/security/data-imports-http-transport.yaml`. The rule bans direct `requests.Session()`,
`requests.<verb>(...)`, and `httpx.Client/AsyncClient/<verb>` inside `sources/**`. Type-only imports
(`from

…(truncated)