PostgreSQL ingest connector.
This guide describes the legacy one-time PostgreSQL ingest endpoint. It copies selected source rows into OriginChainDB. For managed copy and live sync, start with Migrate and check the source options supported by your instance.
- Map source and targetCheck the source table, target schema and primary key.
- Supply permitted source accessUse the documented connection contract for this endpoint.
- Run the one-shot copyInspect the operation’s response and any failed rows.
- Compare the copied recordsVerify IDs, values and counts before switching application reads.
Quickstart.
curl -X POST "https://acme.db.originchain.ai/v1/tenants/$T/ingest/postgres/sync" \
-H "Authorization: Bearer $OC_TOKEN" \
-H "Content-Type: application/json" \
-d '{
"source": {
"host": "pg.internal",
"port": 5432,
"database": "production",
"user": "readonly",
"password": "***",
"sslmode": "require"
},
"mappings": [
{
"source_table": "public.orders",
"target_schema": "orders",
"primary_key": ["id"],
"columns": ["id", "customer_id", "amount", "status", "created_at"]
}
]
}'Source config.
host / port / databaseSource endpoint. Must be reachable from your OriginChainDB instance — public, peered, or via a private endpoint on Enterprise.user / passwordRead-only Postgres role recommended. Credentials are held in memory for the duration of the sync.sslmode"require" or "verify-full". Plaintext is rejected.
Target schema mapping.
Each mapping picks a source table, a target OriginChainDB schema, a primary key, and the columns to pull. The connector creates the OriginChainDB schema if it does not exist, with column types inferred from the Postgres type map. You can declare the schema up-front in the manifest if you want explicit control.
Scheduling.
v1 is one-shot: each POST runs a complete sync. Schedule from your side (cron, a queue, or a tiny scheduler script) until the engine ships a managed scheduler. The endpoint is idempotent on primary key — the same row pulled twice overwrites; no duplicates.
import os, requests, schedule, time
def sync_once():
requests.post(
f"{os.environ['OC_ENDPOINT']}/v1/tenants/{os.environ['T']}/ingest/postgres/sync",
headers={"Authorization": f"Bearer {os.environ['OC_TOKEN']}"},
json={
"source": {...},
"mappings": [...],
},
timeout=900,
).raise_for_status()
schedule.every(15).minutes.do(sync_once)
while True:
schedule.run_pending()
time.sleep(1)What's next.
For ongoing copy and supported live-sync sources, use Migrate. That managed workflow is separate from the one-time endpoint on this page. Confirm source support, access and job status in your deployment before planning a cutover.