# Snowflake connector Source: /docs/connectors/snowflake.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] 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](/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://docs.snowflake.com/en/user-guide/python-connector) to select concrete resources. | Suggested table | One row represents | | --- | --- | | `databases` | a database | | `schemas` | a schema | | `tables` | a table | | `views` | a view | | `query_rows` | a query result row | ### Relationships and extraction considerations [#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 [#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](/docs/reference/cli/overview) for registration and configuration. ### Connection settings used by the example [#connection-settings-used-by-the-example] | Setting | Requirement | Purpose | | --- | --- | --- | | `account` | Required | Account | | `user` | Required | User | | `password` | Required; secret | Password | | `warehouse` | Required | Warehouse | | `database` | Required | Database | | `schema` | Optional | Schema | | `role` | Optional | Role | ### 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 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` [#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` [#connection-schema-configjson] This file contains a JSON Schema, not actual credential values. Provide those through platform configuration. ```json { "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` [#connector-connectorpy] ```python 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` [#sync-schema-syncstable_rowsconfigjson] ```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` [#sync-syncstable_rowssyncpy] ```python 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 [#running-and-extending-the-example] Register the connector before the sync and commit each resource separately: ```sh vortexify create connector snowflake vortexify create sync snowflake/table_rows ``` 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://docs.snowflake.com/en/user-guide/python-connector) — 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.