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.:

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:

   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 — 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):

# 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:

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:

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:

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:

  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

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

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)