Data sources API

These endpoints are what Settings → Data sources uses. See Data sources for the screens, File drops, Kafka topics and Change data capture for how collection works.

Before you start

Who can call them. Signed-in people with the datasources.manage permission (by default, the Administrator role). An API key gets 401 sign in required, and a person without the permission gets 403 missing permission datasources.manage. Every change, test, run and download is written to the audit log.

Signing in with curl. Sign in once and keep the session cookie, as described in Alerts and cases API: before you start, then add -b ~/.sp-cookies to every request below.

Endpoints at a glance

Endpoint	Does	Success
GET /api/v1/data-sources/options	What this installation allows	200
GET /api/v1/data-sources	List data sources with their status	200
POST /api/v1/data-sources	Add one	201 {"id": …}
PUT /api/v1/data-sources/{id}	Replace its settings; credentials left out are kept	204
DELETE /api/v1/data-sources/{id}	Delete it and its file history	204
POST /api/v1/data-sources/{id}/test	Test the connection, loading nothing	200
POST /api/v1/data-sources/{id}/run	Make a file drop due now	202
GET /api/v1/data-sources/{id}/files	A file drop’s file history	200
GET /api/v1/data-source-files/{id}	One file, with its first 200 problems	200
GET /api/v1/data-source-files/{id}/results	Download a file’s results (CSV)	200
GET /api/v1/data-sources/{id}/mapping	A change data capture source’s mapping and its versions	200
PUT /api/v1/data-sources/{id}/mapping	Save a new mapping version (may need approval)	200 {"version": n}, or 202 with the request
GET /api/v1/data-sources/{id}/cdc	Where a change data capture source stands, with the DBA instructions for Postgres	200
POST /api/v1/data-sources/{id}/cdc/resync	Read everything again	202
POST /api/v1/data-sources/{id}/cdc/drop-slot	Drop a Postgres source’s replication slot and pause it	202
GET /api/v1/cdc/templates, GET /api/v1/cdc/fields	Example mappings; the fields each item kind can map	200
POST /api/v1/cdc/test	Try a mapping on sample change events, storing nothing	200

Options

GET /api/v1/data-sources/options

Field	Meaning
topic_prefix	Topics this workspace may read must start with this, such as inst.kweku-pay.. Empty: any topic name
shared_topics	The installation’s shared topics, read with the sp-workspace header
allow_http	Plain-http S3 endpoints are allowed
allow_private	Private-network hosts are allowed
item_kinds	transactions, customers, accounts, account_events, fx_rates
csv_transaction_columns, csv_customer_columns	The CSV columns, as for Settings → Data import

The data source object

Field	Meaning
id, name	Its ID and name (1 to 100 characters, unique in the workspace)
kind	kafka, sftp, s3, cdc (Debezium topics) or postgres_cdc (Postgres replication). Fixed once created
item_kind	transactions, customers, accounts, account_events or fx_rates. Always mapped for cdc and postgres_cdc: each table maps to its own kind
format	json (Kafka); for files auto, csv, jsonl, camt053, pain001; for cdc debezium_json (default) or debezium_avro; for postgres_cdc always pgoutput
config	The settings below. Never contains a secret
has_credentials	Whether a key or secret is stored. The values are never returned
enabled	false while paused
schedule_minutes	File drops: how often it looks for files, 1 to 1,440
next_run_at	File drops, while enabled: when it next looks
running	A connector is working on it now
last_run_at, last_success_at	When it last ran, and last succeeded
last_error, last_error_at	The last problem, and when
files_processed, rows_accepted, rows_rejected	Totals since it was created. Duplicates count as accepted
lag	Kafka: records waiting on the topic
created_at	When it was created

config by kind

Field	Kind	Meaning
topic	Kafka	The topic: letters, digits, ., _ and -, up to 249 characters. Not one of SpeculaPro’s own (sp. by default)
mode	Kafka	topic (the topic is this workspace’s; default) or header (a shared topic, read by the sp-workspace header)
dlq_topic	Kafka	Where rejected records go. Default <topic>.dlq. Never one of SpeculaPro’s own topics or a shared topic, and inside the workspace’s namespace when there is one. See The dead-letter topic
host, port	SFTP	Host name or address, without a port; port 1 to 65535, default 22
username	SFTP	Up to 128 characters
host_key	SFTP	The server’s public key (ssh-ed25519 AAAA…, optionally with host names as in known_hosts) or its SHA256: fingerprint. Required
endpoint	S3	https://host[:port], without credentials or a path. A bare host is read as https://
region	S3	Default us-east-1
bucket	S3	3 to 63 lower-case letters, digits, . and -
path_style	S3	true for MinIO and Ceph
access_key_id	S3	Required
path	Files	SFTP directory or S3 prefix. No .., wildcards or backslashes
pattern	Files	File name pattern, such as *.csv. No /. Default *
results_path	Files	Where results files are written. Empty: not written back. Must differ from path
after	Files	move, tag (S3 only) or leave (default)
processed_path	Files	Required with move, and different from path
retention_days	Files	With move only: delete moved files older than this, 0 to 3,650 (0 keeps them)
topics	cdc	1 to 50 Debezium topics, usually one per table. The same rules as topic: not SpeculaPro’s own, not a shared topic, inside the namespace when there is one
dlq_topic	cdc	Where rejected change events go. Default: the first topic + .dlq. The same rules as for Kafka
schema_registry_url	cdc	Required for debezium_avro: https://host[:port], without credentials (plain http needs SP_CONNECTOR_ALLOW_HTTP)
registry_username	cdc	Schema registry user (basic authentication). Its password goes in credentials.registry_password
host, port	postgres_cdc	The database host, without a port; port default 5432
username	postgres_cdc	The replication user, up to 63 characters. Its password goes in credentials.password
database	postgres_cdc	Required
publication	postgres_cdc	The publication the DBA created: lower-case letters, digits and _
slot	postgres_cdc	Optional replication slot name (default speculapro_ and part of the source’s ID). Can’t change once the slot exists
ssl_mode	postgres_cdc	verify-full (default), verify-ca, require, or disable (only with SP_CONNECTOR_CDC_ALLOW_PLAINTEXT). See TLS
ca_cert	postgres_cdc	PEM certificate of the CA that signed the server’s certificate. Required for verify-ca; default the system’s trusted CAs

Add a data source

POST /api/v1/data-sources

Field	Required	Meaning
name	Yes	1 to 100 characters
kind	Yes	kafka, sftp or s3
item_kind	Yes	What it carries
format	No	Kafka: json (the default). Files: auto (default), jsonl, csv (transactions and customers only), camt053 or pain001 (transactions only)
config	Yes	As above. Settings that don’t apply to the kind are dropped
credentials	SFTP and S3	Write-only: private_key (OpenSSH or PEM) and passphrase for SFTP; secret_access_key for S3. Kafka sources take none: they use the installation’s broker credentials
enabled	No	Default true
schedule_minutes	No	File drops, 1 to 1,440. Default 15

curl -s -b ~/.sp-cookies -H 'Content-Type: application/json' \
  -d '{
    "name": "Core banking statements",
    "kind": "sftp",
    "item_kind": "transactions",
    "format": "camt053",
    "config": {
      "host": "sftp.yourbank.com.gh",
      "username": "speculapro",
      "host_key": "SHA256:<the server fingerprint, 43 characters>",
      "path": "/outbound/speculapro",
      "pattern": "*.xml",
      "results_path": "/outbound/speculapro/results",
      "after": "move",
      "processed_path": "/outbound/speculapro/done",
      "retention_days": 30
    },
    "credentials": {"private_key": "<the OpenSSH private key>"},
    "schedule_minutes": 60
  }' \
  "$SP/api/v1/data-sources"

The private key and secret are encrypted with the workspace’s key before they are stored.

Change, pause or delete

PUT /api/v1/data-sources/{id}

The same body as creating, in full: it replaces the settings. kind must stay the same. Leave credentials out (or empty) to keep the stored ones. To pause, send the same settings with "enabled": false; to resume, true. Pausing a file drop also clears its last error.

DELETE /api/v1/data-sources/{id}

Deletes the data source, its file history and stored results files. Data already loaded stays.

A postgres_cdc source whose replication slot still exists is refused, so no orphaned slot keeps WAL on the institution’s database: drop the slot first. If the DBA has already dropped it, add ?force=true. A forced delete leaves any remaining slot on the institution’s database for the DBA to drop. It is audited as datasource.delete_forced, with the slot’s name, and needs approval when the workspace requires it for Data sources (datasources.connection): send X-Change-Reason and the answer is 202 with the request. force on a source with no slot is an ordinary delete.

Test, run

POST /api/v1/data-sources/{id}/test

Connects with the saved settings and reports what it finds. Nothing is loaded. The answer is always 200, with ok saying whether it worked:

{"ok": true, "message": "Connected. 3 of 4 files in /outbound/speculapro match *.xml."}

{"ok": false, "message": "ssh: handshake failed: host key mismatch: the server presented SHA256:… (ssh-ed25519), but SHA256:… is pinned; if the server's key really changed, check the new one with its administrator and update the host key"}

The test runs on a connector, never on the web tier. When the api and connector roles share a process (all), it runs there; otherwise the request queues it and waits up to 25 seconds for a connector to run it (each test gives up after 20 seconds). If no connector answers, the reply is still 200, with "ok": false and the message No connector answered within 25 seconds. Tests run on the connector role (the only one allowed to reach outside systems): check that it is running and healthy, then try again. It is audited as datasource.test, with the result.

POST /api/v1/data-sources/{id}/run

Makes an enabled file drop due now; a connector picks it up within SP_CONNECTOR_INTERVAL (30 seconds by default). 404 not found: no enabled file drop with that id for a Kafka topic, a paused file drop, or an unknown ID.

File history

GET /api/v1/data-sources/{id}/files

Query parameter	Meaning
limit	1 to 200, default 50

Newest first. An unknown data source gives an empty list.

GET /api/v1/data-source-files/{id}

Field	Meaning
id, name, size	The file
sha256	Its checksum (in the single-file answer only)
format	How it was read: csv, jsonl, camt053, pain001
status	processing, done, retry (loaded again on the next run) or failed
rows, accepted, duplicate, stale, rejected, retry	Row counts
error	Why the whole file failed (including loading this file stopped without finishing 3 times … after three interrupted loads), or a problem moving it or writing its results
errors	The first 200 problems, each {"line": …, "message": …} (in the single-file answer only)
results_name	Where its results file was written back
has_results	A copy of the results is kept for download
attempts	How many times it has been loaded
started_at, finished_at	When

GET /api/v1/data-source-files/{id}/results

The results CSV (line,external_id,status,errors), as text/csv with Content-Disposition: attachment; filename="<file>.results.csv". 404 when no copy was kept. Audited as datasource.results_download. See The results file.

Change data capture

These apply to cdc and postgres_cdc sources. Reading needs datasources.manage or settings.view; everything else needs datasources.manage. See Change data capture for the mapping format.

GET /api/v1/data-sources/{id}/mapping

The active mapping, or ?version=n for an earlier one, and the list of versions.

Field	Meaning
active_version	The version the connector uses; null until one is saved (nothing is read until then)
version, spec, note	The version shown, its mapping and its note
versions	The versions, newest first (up to 100)

PUT /api/v1/data-sources/{id}/mapping

{"spec": {"tables": [ … ]}, "note": "maps the new channel codes"}

note is up to 500 characters. The mapping is checked in full before it is stored (unknown fields, missing required fields, bad conditions, time zones and scales are refused with 422). The new version becomes active at once and a stopped source resumes. When the workspace requires approval for Data source mappings (datasources.mapping, on by default), send X-Change-Reason and the answer is 202 with the request; it applies when approved. See Change requests API. Every saved version is audited as datasource.mapping.

GET /api/v1/data-sources/{id}/cdc

Field	Meaning
phase	new (waiting to start), snapshot (first copy), streaming or failed (stopped)
mapping_version	The active mapping version
error, error_at	Why the source stopped
last_event_at, last_position	When the last applied change was committed at the source, and its log position
events_total, snapshot_rows, snapshot_done_at	Changes applied; rows in the first copy, and when it finished
slot, checkpoint_lsn, drop_requested	Postgres: the replication slot, the position everything before which was accepted, and whether a drop is pending
slot_lag_bytes, slot_checked_at, lag_alert_bytes	Postgres: the WAL the database keeps for the slot, when it was measured, and the alert threshold
recent_errors	The newest 50 rejected changes: table, operation, key, log position and reasons. All of them are in the source’s dead-letter topic
dba	Postgres: the DBA instructions (setup_sql, server_conf, hba, monitor_sql, drop_slot_sql) with this source’s names filled in

POST /api/v1/data-sources/{id}/cdc/resync

Answers 202. Postgres: the connector drops the slot and takes a new first copy of every mapped table. Debezium: the source restarts from its committed offsets (re-snapshot in Debezium, or reset the consumer group, to read a topic from the start). Rows already applied are recognised. Audited as datasource.cdc_resync.

When the workspace requires approval for Data sources (datasources.connection), send X-Change-Reason; the 202 then carries a change request and nothing happens until it is approved. Approval is refused if the source’s settings or slot changed in the meantime. See Change requests API.

POST /api/v1/data-sources/{id}/cdc/drop-slot

Answers 202. Within about a minute a connector drops the Postgres source’s replication slot on the institution’s database and pauses the source. 404 not found: no replication slot recorded for this source if there is none. Audited as datasource.cdc_drop_slot. Gated like a resync when the workspace requires approval for Data sources.

GET /api/v1/cdc/templates and GET /api/v1/cdc/fields

The example mappings, and for each item kind the fields a mapping can set (path, type, required and, for fixed lists, enum).

POST /api/v1/cdc/test

{"spec": {"tables": [ … ]}, "sample": "{\"table\": \"core.txn_journal\", \"op\": \"c\", \"after\": { … }}"}

sample is a Debezium change event, a list of up to 100, or row changes with table, op, before and after. Nothing is stored or sent. The answer has mapping_errors and, for each change, table, op, kind, item (as the ingest API would receive it), skipped, errors, stops (when the source would stop there) and idempotency_key.

Errors

Status	Meaning	Examples
401	Not signed in, or called with an API key	sign in required
403	Missing the permission	missing permission datasources.manage
404	No such data source or file in your workspace	not found; not found: no enabled file drop with that id
422	A field fails its rules. The message starts invalid:	invalid: name is required (up to 100 characters); invalid: a data source with that name already exists; invalid: at most 50 data sources per workspace; invalid: a data source's kind can't change; create a new one; invalid: topic: letters, digits, '.', '_' and '-' only; invalid: sp.txn.ingested is one of SpeculaPro's own topics; invalid: that topic can't be used by this workspace: choose another topic (the topic is already mapped, possibly by another workspace; the message doesn’t say which); invalid: on this deployment, this workspace's topics are named inst.kweku-pay.<name>; invalid: dlq_topic must be named inst.kweku-pay.<name>; invalid: dlq_topic can't be one of SpeculaPro's own topics or a shared topic; invalid: shared.in is a shared topic: use mode header; invalid: dlq_topic must be a different, valid topic name; invalid: Kafka sources use the deployment's broker credentials; invalid: host: destination not allowed: sftp.internal is an internal host; invalid: host: a host name or address only (set the port separately); invalid: host_key: required: the server's host key, so no other server can impersonate it; invalid: credentials.private_key is required (key authentication); invalid: endpoint: use https (plain http needs SP_CONNECTOR_ALLOW_HTTP); invalid: bucket: 3-63 lower-case letters, digits, '.' and '-'; invalid: CSV files carry transactions or customers (the Data import columns); use JSON Lines for accounts; invalid: ISO 20022 files carry transactions; invalid: pattern: a file name pattern like *.csv; invalid: path: no '..', wildcards or backslashes; invalid: tagging is for S3 objects; move or leave SFTP files; invalid: results_path must differ from path, or results files would be read as input; invalid: retention_days applies to moved files only (0-3650). Change data capture: invalid: topics: give 1 to 50 Debezium topics; invalid: core.x is a shared topic: Debezium topics belong to one workspace; invalid: Debezium change events are debezium_json or debezium_avro; invalid: schema_registry_url is required for Avro; invalid: schema_registry_url: use https (plain http needs SP_CONNECTOR_ALLOW_HTTP); invalid: Debezium sources use the deployment's broker credentials; only a schema registry password can be set; invalid: username is required (the replication user); invalid: publication: lower-case letters, digits and '_' (the name the DBA created); invalid: ssl_mode: TLS is required (disable needs SP_CONNECTOR_CDC_ALLOW_PLAINTEXT); invalid: ca_cert is required for verify-ca; invalid: credentials.password is required (the replication user's password); invalid: the replication slot is speculapro_… and can't be renamed: drop it first (Drop slot), or create a new source; invalid: replication slot speculapro_… is on the database this source reads now: drop it first (Drop slot), then change where it connects; invalid: this source's replication slot still exists on your database and keeps WAL there: use Drop slot first … (delete). Mappings: invalid: map at least one table; invalid: core.txn_journal: amount_minor: scale_currency needs the currency field mapped; invalid: core.txn_journal: occurred_at is required for transactions

Related

• Data sources
• File drops
• Kafka topics
• ISO 20022 files
• Change data capture
• Imports API