Docs
Connectors

Snowflake connector

Cloud data warehouse for SQL analytics, tables, views, and data sharing.

View Markdown

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

Cloud data warehouse for SQL analytics, tables, views, and data sharing.

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 tableOne row represents
databasesa database
schemasa schema
tablesa table
viewsa view
query_rowsa query result row

Relationships and extraction considerations

Choose stable object IDs and preserve references between related resources. Nested arrays may need child tables. The listed names are proposed output datasets; validate the actual endpoint, fields, permissions, and pagination before adding each sync.

Connecting to Snowflake

Uses a database driver and a read-only database user; SQL identifiers are quoted, not interpolated unescaped. Discovery reads the configured schema from INFORMATION_SCHEMA. Uses a warehouse and user/password authentication; organizations requiring MFA/SSO should adapt client() to key-pair or OAuth authentication. The driver prefetches server result chunks independently of fetchmany().

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

SettingRequirementPurpose
accountRequiredAccount
userRequiredUser
passwordRequired; secretPassword
warehouseRequiredWarehouse
databaseRequiredDatabase
schemaOptionalSchema
roleOptionalRole

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
snowflake-connector-python>=3,<5
pandas>=2,<3

Use a supported Snowflake Python/OS combination. Native dependency wheels and account authentication requirements must be checked in the target runtime.

Worked example: sync table_rows

table_rows — Reads the configured table with its actual columns; full read, no checkpoint cursor.

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/snowflake/ and sync files in connectors/snowflake/syncs/table_rows/. 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": {
    "account": {
      "type": "string",
      "minLength": 1,
      "title": "Account",
      "order": 0
    },
    "user": {
      "type": "string",
      "minLength": 1,
      "title": "User",
      "order": 1
    },
    "password": {
      "type": "string",
      "minLength": 1,
      "x-vortexify-secret": true,
      "title": "Password",
      "order": 2
    },
    "warehouse": {
      "type": "string",
      "minLength": 1,
      "title": "Warehouse",
      "order": 3
    },
    "database": {
      "type": "string",
      "minLength": 1,
      "title": "Database",
      "order": 4
    },
    "schema": {
      "type": "string",
      "minLength": 1,
      "default": "PUBLIC",
      "title": "Schema",
      "order": 5
    },
    "role": {
      "type": "string",
      "minLength": 1,
      "title": "Role",
      "order": 6
    }
  },
  "required": [
    "account",
    "user",
    "password",
    "warehouse",
    "database"
  ]
}

Connector: connector.py

import snowflake.connector
from contextlib import closing
from vortexify import Column, Connector, DiscoveredTable


class SnowflakeConnector(Connector):
    vendor_domain = "snowflake.com"

    def client(self):
        if self._client is None:
            self._client = snowflake.connector.connect(
                account=self.config["account"],
                user=self.config["user"],
                password=self.config["password"],
                warehouse=self.config["warehouse"],
                database=self.config["database"],
                schema=self.config.get("schema", "PUBLIC"),
                role=self.config.get("role"),
                autocommit=True,
                client_prefetch_threads=1,
            )
        return self._client

    def check(self) -> bool:
        with closing(self.client().cursor()) as cursor:
            cursor.execute("SELECT 1")
            return cursor.fetchone()[0] == 1

    def discover(self) -> list[DiscoveredTable]:
        resources = {}
        with closing(self.client().cursor()) as cursor:
            cursor.execute(
                "SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE, IS_NULLABLE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA=%s ORDER BY TABLE_NAME, ORDINAL_POSITION",
                (self.config.get("schema", "PUBLIC"),),
            )
            for table, name, data_type, nullable in cursor:
                if table not in resources:
                    resources[table] = DiscoveredTable(
                        name=str(self.config.get("schema", "PUBLIC")) + "." + table,
                        schema=[],
                    )
                resources[table].schema.append(
                    Column(name=name, type=data_type, nullable=nullable == "YES")
                )
        return list(resources.values())

    def close(self) -> None:
        if self._client is not None:
            try:
                self._client.close()
            finally:
                self._client = None

Sync schema: syncs/table_rows/config.json

{
  "type": "object",
  "properties": {
    "table": {
      "type": "string",
      "title": "Table",
      "description": "Table name within the configured schema/database.",
      "minLength": 1,
      "order": 0
    }
  },
  "required": [
    "table"
  ]
}

Sync: syncs/table_rows/sync.py

from collections.abc import Iterator
from math import isfinite
import pandas as pd
from vortexify import Sync
from contextlib import closing


class SnowflakeTableRowsSync(Sync):

    def preview(self) -> pd.DataFrame:
        parts = [
            self.connector.config["database"],
            self.connector.config.get("schema", "PUBLIC"),
            self.config["table"],
        ]
        qualified = ".".join(('"' + part.replace('"', '""') + '"' for part in parts))
        with closing(self.client.cursor()) as cursor:
            cursor.execute("SELECT * FROM " + qualified + " LIMIT 100")
            columns = [column[0] 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]:
        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 = max(1, int(target_bytes / max(1.0, avg_row_size_bytes) / 4))
        batch_size = min(batch_size, 10000)
        parts = [
            self.connector.config["database"],
            self.connector.config.get("schema", "PUBLIC"),
            self.config["table"],
        ]
        qualified = ".".join(('"' + part.replace('"', '""') + '"' for part in parts))
        with closing(self.client.cursor()) as cursor:
            cursor.execute("SELECT * FROM " + qualified)
            columns = [column[0] for column in cursor.description]
            while rows := cursor.fetchmany(batch_size):
                yield pd.DataFrame(rows, columns=columns, dtype=object)

Running and extending the example

Register the connector before the sync and commit each resource separately:

vortexify create connector snowflake
vortexify create sync snowflake/table_rows

Edit 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

Snipe-IT connector

IT asset and equipment management for hardware, assignments, licenses, and audits.

Stripe connector

Connect to Stripe payment, billing, customer, and account-balance data.

On this page

What this connector is forData and tables you can pullRelationships and extraction considerationsConnecting to SnowflakeConnection settings used by the examplerequirements.inWorked example: sync table_rowsConnection schema: config.jsonConnector: connector.pySync schema: syncs/table_rows/config.jsonSync: syncs/table_rows/sync.pyRunning and extending the exampleSources and further reading