PostgreSQL

Each row becomes one bucket object. The same connection also backs the sql_lookup retriever stage.

Overview

A PostgreSQL connection serves two paths:

Sync. Rows from a table land in a bucket as JSON objects, with incremental sync via a watermark column.
Lookup. The sql_lookup retriever stage enriches documents mid-pipeline.

Prerequisites

PostgreSQL 12 or later.
A role with CONNECT on the database and USAGE on the schema.
SELECT on every table you sync or query.
Network access from Mixpeek to the server.

Configuration

Connection-level fields

Field	Required	Description
`credentials.username`	Yes	PostgreSQL username
`credentials.password`	Yes	Encrypted at rest, never returned in responses
`host`	Yes	Hostname or IP
`database`	Yes	Database name
`port`	No	Defaults to `5432`
`schema`	No	Defaults to `public`
`ssl_mode`	No	`disable`, `allow`, `prefer` (default), `require`, `verify-ca`, `verify-full`
`incremental_column`	No	`TIMESTAMP` or `DATE` column used as the sync watermark
`primary_key_columns`	No	Columns that form the key, used to build stable object IDs
`query_timeout_seconds`	No	Defaults to `300`
`fetch_size`	No	Rows per batch, defaults to `1000`

Sync-level fields

Field	Required	Description
`source_path`	Yes	`table`, or `schema.table` to override the connection schema
`polling_interval_seconds`	No	Defaults to `300`

source_path accepts one or two dot-separated parts. Three or more returns a validation error.

Setup

Create a read-only role

CREATE ROLE mixpeek_sync LOGIN PASSWORD 'choose-a-strong-password';
GRANT CONNECT ON DATABASE production TO mixpeek_sync;
GRANT USAGE ON SCHEMA public TO mixpeek_sync;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mixpeek_sync;

Create the storage connection

from mixpeek import Mixpeek

client = Mixpeek(api_key="your-mixpeek-api-key")

connection = client.organizations.connections.create(
    name="Production Postgres",
    provider_type="postgresql",
    provider_config={
        "credentials": {
            "type": "username_password",
            "username": "mixpeek_sync",
            "password": "choose-a-strong-password",
        },
        "host": "db.example.com",
        "port": 5432,
        "database": "production",
        "schema": "public",
        "ssl_mode": "require",
        "incremental_column": "updated_at",
        "primary_key_columns": ["id"],
    },
)
print(f"Created connection: {connection['connection_id']}")

import { Mixpeek } from 'mixpeek-sdk'

const client = new Mixpeek({ apiKey: 'your-mixpeek-api-key' })

const connection = await client.organizations.connections.create({
  name: 'Production Postgres',
  provider_type: 'postgresql',
  provider_config: {
    credentials: {
      type: 'username_password',
      username: 'mixpeek_sync',
      password: 'choose-a-strong-password',
    },
    host: 'db.example.com',
    port: 5432,
    database: 'production',
    schema: 'public',
    ssl_mode: 'require',
    incremental_column: 'updated_at',
    primary_key_columns: ['id'],
  },
})
console.log('Created connection:', connection.connection_id)

curl -X POST https://api.mixpeek.com/v1/organizations/connections \
  -H "Authorization: Bearer YOUR_MIXPEEK_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Production Postgres",
    "provider_type": "postgresql",
    "provider_config": {
      "credentials": {
        "type": "username_password",
        "username": "mixpeek_sync",
        "password": "choose-a-strong-password"
      },
      "host": "db.example.com",
      "port": 5432,
      "database": "production",
      "schema": "public",
      "ssl_mode": "require",
      "incremental_column": "updated_at",
      "primary_key_columns": ["id"]
    }
  }'

Sync a table

sync = client.buckets.syncs.create(
    bucket_id="bkt_your_bucket_id",
    connection_id=connection["connection_id"],
    source_path="customers",
    polling_interval_seconds=1800,
)
print(f"Sync created: {sync['sync_config_id']}")

curl -X POST https://api.mixpeek.com/v1/buckets/bkt_your_bucket_id/syncs \
  -H "Authorization: Bearer YOUR_MIXPEEK_API_KEY" \
  -H "X-Namespace: ns_your_namespace_id" \
  -H "Content-Type: application/json" \
  -d '{
    "connection_id": "conn_your_connection_id",
    "source_path": "analytics.customers",
    "polling_interval_seconds": 1800
  }'

Incremental sync

Set incremental_column to a TIMESTAMP or DATE column. Mixpeek records the highest value it has read and filters the next run to rows above it. Set primary_key_columns so object IDs stay stable across runs. Without it, a re-read can create a second object for the same row.

SQL Lookup stage

Enrich documents from a SQL source inside a retriever.

Snowflake

Warehouse connection with the same two paths.

Get started

Connect your data

Extract features

Build retrievers

Enrich & organize

Integrate & operate

Resources

Overview

Prerequisites

Configuration

Connection-level fields

Sync-level fields

Setup

Incremental sync

SQL Lookup stage

Snowflake

​Overview

​Prerequisites

​Configuration

​Connection-level fields

​Sync-level fields

​Setup

​Incremental sync

​Related

SQL Lookup stage

Snowflake

Overview

Prerequisites

Configuration

Connection-level fields

Sync-level fields

Setup

Incremental sync

Related