Sources are how a tenant gets its own data into Causeloop. services/api/sources_api.py owns registering connectors, testing/syncing them, and per-source file upload; every mutating route requires sources.manage or ingest.run (require_permission), and credentials are write-only — encrypted immediately on submission (services/api/secrets.py) and never returned in any response afterward, only a last-4-style credential_hint persists for display.
Connector kinds
Four connector kinds are registered in services/ingest/connectors/registry.py’s CONNECTOR_REGISTRY, each implementing the same test_connection() / pull() / pull_file() contract (services/ingest/connectors/base.py) so the ingestion pipeline never needs to know which kind it’s talking to:
GET /sources/connectors returns {kind: schema} for every registered kind, each schema being that connector’s Pydantic config_schema.model_json_schema() — the intent (per the module’s own docstring) is for a source-creation form to be generated directly from this schema rather than hand-built per kind.
GET /sources/connectors, POST /sources/{source_id}/test-connection, POST /sources/{source_id}/sync, GET /sources/{source_id}/history, and GET /sources/jobs/{job_id}/events are all implemented in services/api/sources_api.py, but services/api/product_app.py’s SOURCE_OPERATIONS allowlist — the actual deployed product surface — admits only GET /sources, POST /sources, and POST /sources/{source_id}/upload. The shipped console (SourcesClient.tsx) matches the allowlist exactly: it lists, creates, and uploads to sources, and never calls connector discovery, test-connection, sync, history, or the job event stream. Those remain router-level capabilities for a future connector-management UI, not stubs returning errors — they simply aren’t wired to any deployed route or frontend surface today.
Source health
Source.health_status (services/api/db/models.py) defaults to "unknown" and is otherwise only ever written to "healthy" or "error" — by test_source_connection() (from the connector’s own test_connection() result) or by sync_source() ("healthy" on a clean sync, "error" with the exception text in health_detail on failure). The frontend’s sourceStatusTone() (SourcesClient.tsx) also maps a "syncing" status to an amber chip defensively, but nothing in the current backend ever sets that value — a source card in the console today only ever shows unknown (neutral), healthy (green), or error (red).
Permissions summary
Registering and using a source
GET /sources
Returns every Source row for the tenant as a SourceSummary: id, kind, name, config, credential_hint, health_status, health_detail, last_synced_at. No credentials, ever.
POST /sources
CreateSourceRequest is {kind, name, config, credentials?}. The handler validates kind against CONNECTOR_REGISTRY (422 if unknown) and validates config against that kind’s config_schema (422 with the Pydantic validation error if it doesn’t fit). If credentials is supplied, it’s JSON-encoded and passed through encrypt_secret(); the stored credential_hint is derived from password, then secret_access_key, then whatever the first credential value is — never the raw secret. Requires sources.manage + CSRF.
POST /sources/{source_id}/upload
The route the shipped console’s “Upload and ingest” button calls. file_upload-kind sources have no remote endpoint to sync, so this pushes the uploaded bytes through the same connector interface (FileUploadConnector.pull_file()) and the same shared pipeline write path every other connector’s sync uses — producing a real dataset version and ingest_jobs row exactly like a scheduled s3/database/kafka sync would. Returns 409 if the target source isn’t kind == "file_upload", 413 if the upload exceeds the object-store size limit, 422 if the file fails to parse. Requires ingest.run + CSRF.
The frontend’s SourcesClient.tsx doesn’t make the user create a source first: handleUpload() looks for an existing file_upload-kind source and creates one named “File uploads” on the fly if none exists, then uploads to it.
Router-level (not deployed): connector discovery, test-connection, sync, history, job events
For completeness — these exist in services/api/sources_api.py and are exercised by tests, but are not part of the current product allowlist:
GET /sources/connectors — the schema-driven form data described above.POST /sources/{source_id}/test-connection — calls the connector’s test_connection() and writes the result back to health_status/health_detail.POST /sources/{source_id}/sync — pulls new records since the source’s last checkpoint (any non-file_upload kind) through the same unified pipeline as upload.GET /sources/{source_id}/history — every IngestJob for that source, newest first (status, stage, progress, dataset_id, error_detail).GET /sources/jobs/{job_id}/events — a Server-Sent Events stream that polls the durable ingest_jobs row (roughly once per second, up to ~5 minutes) and pushes one data: line per change, closing once the job reaches a terminal status (succeeded/failed/cancelled). Polling the durable Postgres record deliberately avoids standing up a second pub/sub authority just for this.
Seeded source kinds (staff onboarding)
POST /admin/tenants/{tenant_id}/sources/seed (services/api/onboarding_api.py, staff/employee-authenticated, part of ONBOARDING_OPERATIONS) pre-registers placeholder Source rows — kind plus a default name, no credentials — for the onboarding wizard’s “Sources” step, where a client’s admin picks which connector kinds they expect to use on day one:
Requesting an unknown kind returns 422 up front so a typo doesn’t silently create nothing. This only saves the client’s admin from re-picking the kind and typing a name for each source they already told onboarding they wanted — actual source registration (the client-facing add-source flow, credentials included) is exactly POST /sources above; there is no second source store. See Client onboarding pipeline for where this step sits in the overall onboarding flow.
How sources feed ingest and training
Every successful upload or sync writes a new Dataset row (tenant_id, source_id, version — versions are strictly increasing per tenant, locked via lock_tenant_dataset_versions() to avoid a race between concurrent syncs) and a batch of DatasetRecord rows, deduplicated on UNIQUE(tenant_id, dataset_id, content_hash) so re-ingesting the same content is a no-op rather than a duplicate. The write path then atomically enqueues a materialize_insights platform job for that new (dataset_id, version) pair — see Insights console for exactly what that job computes and when the resulting snapshot becomes visible to the tenant console. A source’s checkpoint column (opaque, per-connector-kind — a content hash for file_upload, an S3 object timestamp, a Kafka consumer-group offset) is what lets a subsequent sync pull only what’s new.
Training a model checkpoint against a dataset version is a separate, staff-driven step (POST /admin/tenants/{tenant_id}/train) — sources and ingest only get data into the tenant schema as dataset versions; they don’t themselves trigger model training.
