# PostgreSQL connector Source: /docs/connectors/postgresql.md Use this as an implementation starting point: adapt the worked example to the requested dataset and the source documentation. See [Custom API](/docs/connectors/custom-api) for the shared SDK pattern and document-upload workflow. ## What this connector is for [#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](/docs/reference/python/connectors-and-syncs). ## Data and tables you can pull [#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](https://www.postgresql.org/docs/current/) 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 [#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 [#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](/docs/reference/cli/overview) for registration and configuration. ### Connection settings used by the example [#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 [#requirementsin] 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. ```text vortexify==0.1.1 psycopg[binary]>=3.2,<4 pandas>=2,<3 ``` `psycopg[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` [#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` [#connection-schema-configjson] This file contains a JSON Schema, not actual credential values. Provide those through platform configuration. ```json { "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` [#connector-connectorpy] ```python 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 = None ``` ### Sync schema: `syncs/snapshot/config.json` [#sync-schema-syncssnapshotconfigjson] ```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` [#sync-syncssnapshotsyncpy] ```python 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 dataframe ``` ### Running and extending the example [#running-and-extending-the-example] Register the connector before the sync and commit each resource separately: ```sh vortexify create connector postgresql vortexify create sync postgresql/snapshot ``` Edit the files, inspect `vortexify status`, and use the [CLI guide](/docs/reference/cli/overview) 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 [#sources-and-further-reading] - [Vendor API and data reference](https://www.postgresql.org/docs/current/) — Resource definitions, authentication, fields, and source-specific limits. Check the selected resource's reference when extending the example. - [Vortexify connector and sync contracts](/docs/reference/python/connectors-and-syncs) — Client ownership, discovery, preview isolation, and DataFrame extraction. - [Vortexify CLI workflow](/docs/reference/cli/overview) — Project paths, resource registration, source commits, and credential configuration.