PostgreSQL connector
Relational database for schema inspection, SQL queries, and incremental extraction.
Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See Custom API for the shared SDK pattern and document-upload workflow.
What this connector is for
Relational database for schema inspection, SQL queries, and incremental extraction.
A connector owns access to the source. Separate syncs shape selected resources into tables for analysis and downstream transformations using the Python SDK.
Data and tables you can pull
These are representative dataset families to consider, not a claim that the worked example implements every resource. Access, fields, and availability depend on the source account, product edition, and permissions. Use the vendor reference to select concrete resources.
| Suggested table | One row represents |
|---|---|
tables | a table |
views | a view |
columns | a column |
snapshot_rows | a source row in a snapshot |
incremental_rows | a source row selected by a checkpoint |
Relationships and extraction considerations
Tables and views depend on the connected database and role; these are resource families, not fixed business tables. Preserve primary keys and types. The snapshot example uses a server-side cursor. A polling high-water mark is a separate design from CDC and does not capture every deletion or late commit.
Connecting to PostgreSQL
PostgreSQLConnector creates and caches a Psycopg connection using host,
database, username, and password. Optional port defaults to 5432 and
sslmode to prefer. Connection timeout is 10 seconds; extraction transactions
are read-only. check() executes SELECT 1. discover() returns readable,
non-system tables and views with column types and nullability.
Use the connector as a context manager, or call close() when finished. Use a
separate connector instance for each concurrent run.
Additional resources may require additional scopes. Connection settings belong to the connector; dataset filters and resource selection belong to each sync. See the CLI workflow for registration and configuration.
Connection settings used by the example
| Setting | Requirement | Purpose |
|---|---|---|
host | Required | Hostname or IP address of the PostgreSQL server. |
port | Optional | Port used by the PostgreSQL server. |
database | Required | Name of the PostgreSQL database to connect to. |
username | Required | Username used to authenticate with PostgreSQL. |
password | Required; secret | Password used to authenticate with PostgreSQL. |
sslmode | Optional | PostgreSQL TLS policy. Use verify-full with a trusted server certificate for verified TLS. |
requirements.in
Save these dependencies in requirements.in alongside connector.py; the example includes packages used by its sync. Install the published vortexify package for local development. Keep sync-specific dependencies in the sync's own requirements.in when splitting the implementation into separate resources.
vortexify==0.1.1
psycopg[binary]>=3.2,<4
pandas>=2,<3psycopg[binary] supplies the PostgreSQL client implementation in supported binary wheels. Check wheel availability for the deployment architecture; source builds have separate libpq/build requirements.
Worked example: sync snapshot
The selected sync reads the resource shown below.
The complete files below implement only this example resource. They illustrate the current SDK signatures, a separate preview, bounded DataFrame output, and source-specific extraction. They do not establish complete source coverage. Live credentials and platform ingestion have not been used to validate this guide.
With default paths, place the connector files in connectors/postgresql/ and sync files in connectors/postgresql/syncs/snapshot/. Use your project configuration if it specifies different paths.
Connection schema: config.json
This file contains a JSON Schema, not actual credential values. Provide those through platform configuration.
{
"type": "object",
"properties": {
"host": {
"title": "Host",
"description": "Hostname or IP address of the PostgreSQL server.",
"examples": [
"database.example.com"
],
"order": 0,
"type": "string"
},
"port": {
"title": "Port",
"description": "Port used by the PostgreSQL server.",
"default": 5432,
"examples": [
5432
],
"order": 1,
"type": "integer",
"minimum": 1,
"maximum": 65535
},
"database": {
"title": "Database",
"description": "Name of the PostgreSQL database to connect to.",
"examples": [
"production"
],
"order": 2,
"type": "string"
},
"username": {
"title": "Username",
"description": "Username used to authenticate with PostgreSQL.",
"examples": [
"postgres"
],
"order": 3,
"type": "string"
},
"password": {
"title": "Password",
"description": "Password used to authenticate with PostgreSQL.",
"order": 4,
"type": "string",
"x-vortexify-secret": true
},
"sslmode": {
"title": "SSL Mode",
"description": "PostgreSQL TLS policy. Use verify-full with a trusted server certificate for verified TLS.",
"type": "string",
"enum": [
"disable",
"allow",
"prefer",
"require",
"verify-ca",
"verify-full"
],
"default": "prefer",
"order": 5
}
},
"required": [
"host",
"database",
"username",
"password"
]
}Connector: connector.py
import psycopg
from psycopg import sql
from psycopg.rows import dict_row
from vortexify import Column, Connector, DiscoveredTable
class PostgreSQLConnector(Connector):
vendor_domain = "postgresql.org"
"Connect to PostgreSQL and discover readable tables and views."
def client(self) -> psycopg.Connection:
if self._client is None or self._client.closed:
self._client = psycopg.connect(
host=self.config["host"],
port=self.config.get("port", 5432),
dbname=self.config["database"],
user=self.config["username"],
password=self.config["password"],
sslmode=self.config.get("sslmode", "prefer"),
connect_timeout=10,
autocommit=True,
)
self._client.read_only = True
return self._client
def check(self) -> bool:
with self.client().cursor() as cursor:
cursor.execute("SELECT 1")
return cursor.fetchone()[0] == 1
def discover(self) -> list[DiscoveredTable]:
with self.client().cursor(row_factory=dict_row) as cursor:
cursor.execute(
"\n SELECT c.table_schema, c.table_name, t.table_type,\n c.column_name, c.data_type, c.is_nullable\n FROM information_schema.columns AS c\n JOIN information_schema.tables AS t\n ON t.table_schema = c.table_schema\n AND t.table_name = c.table_name\n WHERE c.table_schema NOT IN ('pg_catalog', 'information_schema')\n AND c.table_schema NOT LIKE 'pg_toast%'\n AND c.table_schema NOT LIKE 'pg_temp_%'\n AND has_table_privilege(\n format('%I.%I', c.table_schema, c.table_name), 'SELECT'\n )\n ORDER BY c.table_schema, c.table_name, c.ordinal_position\n "
)
resources = {}
for column in cursor:
key = (column["table_schema"], column["table_name"])
resource = resources.setdefault(
key,
DiscoveredTable(name=sql.Identifier(*key).as_string(), schema=[]),
)
resource.schema.append(
Column(
name=column["column_name"],
type=column["data_type"],
nullable=column["is_nullable"] == "YES",
)
)
return list(resources.values())
def close(self) -> None:
"""Release the cached connection when the caller finishes using it."""
if self._client is not None:
try:
self._client.close()
finally:
self._client = NoneSync schema: syncs/snapshot/config.json
{
"type": "object",
"properties": {
"schema": {
"title": "Schema",
"description": "Schema containing the table to sync.",
"default": "public",
"examples": [
"public"
],
"order": 0,
"type": "string",
"minLength": 1
},
"table": {
"title": "Table",
"description": "Name of the PostgreSQL table to sync.",
"examples": [
"orders"
],
"order": 1,
"type": "string",
"minLength": 1
}
},
"required": [
"schema",
"table"
]
}Sync: syncs/snapshot/sync.py
from collections.abc import Iterator
from math import isfinite
from uuid import uuid4
import pandas as pd
from psycopg import sql
from vortexify import Sync
class PostgreSQLSnapshotSync(Sync):
"""Read the complete table using a single SELECT and bounded batches."""
def preview(self) -> pd.DataFrame:
query = sql.SQL("SELECT * FROM {} LIMIT 100").format(
sql.Identifier(self.config.get("schema", "public"), self.config["table"])
)
with self.client.transaction():
with self.client.cursor() as cursor:
cursor.execute(query)
columns = [column.name for column in cursor.description]
return pd.DataFrame(cursor.fetchall(), columns=columns, dtype=object)
def extract(
self, *, avg_row_size_bytes: float | None = None
) -> Iterator[pd.DataFrame]:
max_batch_size = 10000
memory_mib = self.compute.memory if self.compute else None
memory_mib = memory_mib if memory_mib is not None and memory_mib > 0 else 1024
target_bytes = min(16 * 1024 * 1024, max(1, memory_mib * 1024 * 1024 // 20))
batch_size = 100
if avg_row_size_bytes is not None:
if not isfinite(avg_row_size_bytes) or avg_row_size_bytes <= 0:
raise ValueError("avg_row_size_bytes must be finite and positive")
batch_size = min(
max_batch_size,
max(1, int(target_bytes / max(1.0, avg_row_size_bytes) / 4)),
)
else:
batch_size = min(max_batch_size, 100)
query = sql.SQL("SELECT * FROM {}").format(
sql.Identifier(self.config.get("schema", "public"), self.config["table"])
)
with self.client.transaction():
with self.client.cursor(name=f"snapshot_{uuid4().hex}") as cursor:
cursor.execute(query)
columns = [column.name for column in cursor.description]
while rows := cursor.fetchmany(batch_size):
dataframe = pd.DataFrame(rows, columns=columns, dtype=object)
observed_row_bytes = dataframe.memory_usage(
index=True, deep=True
).sum() / len(dataframe)
row_bytes = max(1.0, avg_row_size_bytes or 0, observed_row_bytes)
batch_size = min(
max_batch_size, max(1, int(target_bytes / row_bytes / 4))
)
yield dataframeRunning and extending the example
Register the connector before the sync and commit each resource separately:
vortexify create connector postgresql
vortexify create sync postgresql/snapshotEdit the files, inspect vortexify status, and use the CLI guide to commit and configure them. Registration alone does not run extraction or deploy the connector.
Preview must run with a fresh connector and disposable state. The extraction implementation determines full/incremental behavior; merely setting a mode does not implement change capture. API/driver failures propagate unless explicitly handled in the code. Check source limits and single-record/response sizes as well as output batch size.
Sources and further reading
- Vendor API and data reference — Resource definitions, authentication, fields, and source-specific limits. Check the selected resource's reference when extending the example.
- Vortexify connector and sync contracts — Client ownership, discovery, preview isolation, and DataFrame extraction.
- Vortexify CLI workflow — Project paths, resource registration, source commits, and credential configuration.