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