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