Ibis Connectors

SnowflakeConnector manages input (read) and output (write) connections to Snowflake using Ibis. Parameters may be specified with environment variables of the same name or through the init() method interface. Variables passed through init() take precedence.

Attributes:

Name	Type	Description
SNOWFLAKE_USER		Snowflake user name.
SNOWFLAKE_ACCOUNT		Snowflake account identifier.
SNOWFLAKE_WAREHOUSE		Snowflake warehouse name.
SNOWFLAKE_ROLE		Snowflake role name.
SNOWFLAKE_PASSWORD		Snowflake password. If not specified, will attempt to authenticate with externalbrowser.
SNOWFLAKE_PKEY		Snowflake private key for key=pair authentification. Private key has precedence over password authetification.
SNOWFLAKE_SOURCE_DATABASE		Snowflake source database name. Use a fully qualified database name (in snowflake terminology DATABASE.SCHEMA; ibis calls this a "database").
SNOWFLAKE_DEST_DATABASE		Snowflake destination database name. Use a fully qualified database name (in snowflake terminology DATABASE.SCHEMA; ibis calls this a "database").

Methods:

Name	Description
connect_dest	Establishes and returns an Ibis backend connection to the destination Snowflake database and schema.
connect_source	Establishes and returns an Ibis backend connection to the source Snowflake database and schema.
get_source_table	str) -> Table: Retrieves a table from the source Snowflake database.
get_dest_table	str) -> Table: Retrieves a table from the destination Snowflake database.
create_view	Table, name_table: Optional[str] = None, overwrite: bool = False) -> View: Create a view of a table in the destination Snowflake database.
create_table	Table, name_table: Optional[str] = None, overwrite: bool = False) -> Table: Materialize a table in the destination Snowflake database.
drop_table	str) -> None: Drop a table from the destination Snowflake database.
drop_view	str) -> None: Drop a view from the destination Snowflake database.

Source code in phenex/ibis_connect.py

class SnowflakeConnector:
    """
    SnowflakeConnector manages input (read) and output (write) connections to Snowflake using Ibis. Parameters may be specified with environment variables of the same name or through the __init__() method interface. Variables passed through __init__() take precedence.

    Attributes:
        SNOWFLAKE_USER: Snowflake user name.
        SNOWFLAKE_ACCOUNT: Snowflake account identifier.
        SNOWFLAKE_WAREHOUSE: Snowflake warehouse name.
        SNOWFLAKE_ROLE: Snowflake role name.
        SNOWFLAKE_PASSWORD: Snowflake password. If not specified, will attempt to authenticate with externalbrowser.
        SNOWFLAKE_PKEY: Snowflake private key for key=pair authentification. Private key has precedence over password authetification.
        SNOWFLAKE_SOURCE_DATABASE: Snowflake source database name. Use a fully qualified database name (in snowflake terminology DATABASE.SCHEMA; ibis calls this a "database").
        SNOWFLAKE_DEST_DATABASE: Snowflake destination database name. Use a fully qualified database name (in snowflake terminology DATABASE.SCHEMA; ibis calls this a "database").

    Methods:
        connect_dest() -> BaseBackend:
            Establishes and returns an Ibis backend connection to the destination Snowflake database and schema.

        connect_source() -> BaseBackend:
            Establishes and returns an Ibis backend connection to the source Snowflake database and schema.

        get_source_table(name_table: str) -> Table:
            Retrieves a table from the source Snowflake database.

        get_dest_table(name_table: str) -> Table:
            Retrieves a table from the destination Snowflake database.

        create_view(table: Table, name_table: Optional[str] = None, overwrite: bool = False) -> View:
            Create a view of a table in the destination Snowflake database.

        create_table(table: Table, name_table: Optional[str] = None, overwrite: bool = False) -> Table:
            Materialize a table in the destination Snowflake database.

        drop_table(name_table: str) -> None:
            Drop a table from the destination Snowflake database.

        drop_view(name_table: str) -> None:
            Drop a view from the destination Snowflake database.
    """

    def __init__(
        self,
        SNOWFLAKE_USER: Optional[str] = None,
        SNOWFLAKE_ACCOUNT: Optional[str] = None,
        SNOWFLAKE_WAREHOUSE: Optional[str] = None,
        SNOWFLAKE_ROLE: Optional[str] = None,
        SNOWFLAKE_PASSWORD: Optional[str] = None,
        SNOWFLAKE_PKEY: Optional[str] = None,
        SNOWFLAKE_SOURCE_DATABASE: Optional[str] = None,
        SNOWFLAKE_DEST_DATABASE: Optional[str] = None,
    ):
        self.SNOWFLAKE_USER = SNOWFLAKE_USER or os.environ.get("SNOWFLAKE_USER")
        self.SNOWFLAKE_ACCOUNT = SNOWFLAKE_ACCOUNT or os.environ.get(
            "SNOWFLAKE_ACCOUNT"
        )
        self.SNOWFLAKE_WAREHOUSE = SNOWFLAKE_WAREHOUSE or os.environ.get(
            "SNOWFLAKE_WAREHOUSE"
        )
        self.SNOWFLAKE_ROLE = SNOWFLAKE_ROLE or os.environ.get("SNOWFLAKE_ROLE")
        self.SNOWFLAKE_PASSWORD = SNOWFLAKE_PASSWORD or os.environ.get(
            "SNOWFLAKE_PASSWORD"
        )
        self.SNOWFLAKE_PKEY = SNOWFLAKE_PKEY or os.environ.get("SNOWFLAKE_PKEY")
        source_db = SNOWFLAKE_SOURCE_DATABASE or os.environ.get(
            "SNOWFLAKE_SOURCE_DATABASE"
        )
        self.SNOWFLAKE_SOURCE_DATABASE = source_db.upper() if source_db else None
        dest_db = SNOWFLAKE_DEST_DATABASE or os.environ.get("SNOWFLAKE_DEST_DATABASE")
        self.SNOWFLAKE_DEST_DATABASE = dest_db.upper() if dest_db else None

        try:
            _, _ = self.SNOWFLAKE_SOURCE_DATABASE.split(".")
        except:
            raise ValueError(
                "Use a fully qualified database name (e.g. CATALOG.DATABASE)."
            )
        if self.SNOWFLAKE_DEST_DATABASE:
            try:
                _, _ = self.SNOWFLAKE_DEST_DATABASE.split(".")
            except:
                raise ValueError(
                    "Use a fully qualified database name (e.g. CATALOG.DATABASE)."
                )

        # Managed credentials: when no user/password/key is given, the environment
        # (e.g. a VS Code session) supplies credentials to snowflake.connector.
        self._managed = not (
            self.SNOWFLAKE_USER or self.SNOWFLAKE_PASSWORD or self.SNOWFLAKE_PKEY
        )
        if self._managed:
            required_vars = ["SNOWFLAKE_SOURCE_DATABASE"]
        else:
            required_vars = [
                "SNOWFLAKE_USER",
                "SNOWFLAKE_ACCOUNT",
                "SNOWFLAKE_WAREHOUSE",
                "SNOWFLAKE_ROLE",
                "SNOWFLAKE_SOURCE_DATABASE",
            ]
        self._check_env_vars(required_vars)
        self._check_source_dest()
        # Source and destination live in the same Snowflake account, so a single
        # backend connection can fully-qualify either database/schema. Sharing one
        # connection (instead of opening separate source/dest connections) means
        # source and dest tables belong to the same Ibis backend and can be joined
        # server-side. Ibis refuses to join tables that come from two different
        # connection objects, even when they point at the same account.
        self.source_connection = self.connect_source()
        self.dest_connection = (
            self.source_connection if self.SNOWFLAKE_DEST_DATABASE else None
        )

    def _check_env_vars(self, required_vars: List[str]):
        for var in required_vars:
            if not getattr(self, var):
                raise ValueError(
                    f"Missing required variable: {var}. Set in the environment or pass through __init__()."
                )

    def _check_source_dest(self):
        if self.SNOWFLAKE_DEST_DATABASE and (
            self.SNOWFLAKE_SOURCE_DATABASE == self.SNOWFLAKE_DEST_DATABASE
            and self.SNOWFLAKE_SOURCE_SCHEMA == self.SNOWFLAKE_DEST_SCHEMA
        ):
            raise ValueError("Source and destination locations cannot be the same.")

    def _load_private_key_bytes(self, pkey_value: str) -> bytes:
        try:
            from cryptography.hazmat.primitives import serialization
        except ImportError as e:
            raise ImportError(
                "Key-pair auth (SNOWFLAKE_PKEY) requires the 'cryptography' package."
            ) from e

        if os.path.isfile(pkey_value):
            with open(pkey_value, "rb") as f:
                raw = f.read()
        else:
            candidate = "".join(pkey_value.split())
            try:
                raw = base64.b64decode(candidate, validate=True)
            except Exception as e:
                raise ValueError(
                    "SNOWFLAKE_PKEY must be either an existing file path or a base64-encoded PEM private key."
                ) from e

        if raw.lstrip().startswith(b"-----BEGIN"):
            key = serialization.load_pem_private_key(raw, password=None)
        else:
            key = serialization.load_der_private_key(raw, password=None)
        return key.private_bytes(
            encoding=serialization.Encoding.DER,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        )

    def _connect(self, database) -> BaseBackend:
        """
        Private method to get a database connection. End users should use connect_source() and connect_dest() to get connections to source and destination databases.
        """
        database, schema = database.split(".")
        #
        # In Ibis speak: catalog = collection of databases
        #                database = collection of tables
        #                schema = columns and column types
        # In snowflake speak: database = collection of schemas = ibis catalog
        #                schema = collection of tables = ibis database
        #
        # In the below connect method, the arguments are the SNOWFLAKE terms.
        #

        # ibis creates JS helper UDFs at connect time, which restricted roles cannot
        # and PhenEx never uses. PHENEX_SNOWFLAKE_UDFS=1 turns them back on.
        create_object_udfs = os.environ.get("PHENEX_SNOWFLAKE_UDFS") == "1"

        if self._managed:
            # connect() with NO arguments makes snowflake.connector load the session's
            # managed credentials; any argument (even database) disables that.
            import snowflake.connector

            sf_con = snowflake.connector.connect()
            with sf_con.cursor() as cur:
                if self.SNOWFLAKE_ROLE:
                    cur.execute(f'USE ROLE "{self.SNOWFLAKE_ROLE}"')
                if self.SNOWFLAKE_WAREHOUSE:
                    cur.execute(f'USE WAREHOUSE "{self.SNOWFLAKE_WAREHOUSE}"')
                cur.execute(f'USE SCHEMA "{database}"."{schema}"')
            return ibis.snowflake.from_connection(
                sf_con, create_object_udfs=create_object_udfs
            )
        elif self.SNOWFLAKE_PKEY:
            return ibis.snowflake.connect(
                user=self.SNOWFLAKE_USER,
                account=self.SNOWFLAKE_ACCOUNT,
                warehouse=self.SNOWFLAKE_WAREHOUSE,
                role=self.SNOWFLAKE_ROLE,
                database=database,
                schema=schema,
                private_key=self._load_private_key_bytes(self.SNOWFLAKE_PKEY),
                create_object_udfs=create_object_udfs,
            )
        elif self.SNOWFLAKE_PASSWORD:
            return ibis.snowflake.connect(
                user=self.SNOWFLAKE_USER,
                password=self.SNOWFLAKE_PASSWORD,
                account=self.SNOWFLAKE_ACCOUNT,
                warehouse=self.SNOWFLAKE_WAREHOUSE,
                role=self.SNOWFLAKE_ROLE,
                database=database,
                schema=schema,
                create_object_udfs=create_object_udfs,
            )
        else:
            return ibis.snowflake.connect(
                user=self.SNOWFLAKE_USER,
                authenticator="externalbrowser",
                account=self.SNOWFLAKE_ACCOUNT,
                warehouse=self.SNOWFLAKE_WAREHOUSE,
                role=self.SNOWFLAKE_ROLE,
                database=database,
                schema=schema,
                create_object_udfs=create_object_udfs,
                # Caches the SSO ID token in the OS keychain so later connects skip the browser.
                client_store_temporary_credential=True,
            )

    def connect_dest(self) -> BaseBackend:
        """
        Establishes and returns an Ibis backend connection to the destination Snowflake database.

        Returns:
            BaseBackend: Ibis backend connection to the destination Snowflake database.
        """
        return self._connect(
            database=self.SNOWFLAKE_DEST_DATABASE,
        )

    def connect_source(self) -> BaseBackend:
        """
        Establishes and returns an Ibis backend connection to the source Snowflake database.

        Returns:
            BaseBackend: Ibis backend connection to the source Snowflake database.
        """
        return self._connect(
            database=self.SNOWFLAKE_SOURCE_DATABASE,
        )

    def get_source_table(self, name_table):
        """
        Retrieves a table from the source Snowflake database.

        Args:
            name_table (str): Name of the table to retrieve.

        Returns:
            Table: Ibis table object from the source Snowflake database.
        """
        return self.source_connection.table(
            name_table, database=self.SNOWFLAKE_SOURCE_DATABASE
        )

    def get_dest_table(self, name_table):
        """
        Retrieves a table from the destination Snowflake database.

        Args:
            name_table (str): Name of the table to retrieve.

        Returns:
            Table: Ibis table object from the destination Snowflake database.
        """
        if self.SNOWFLAKE_DEST_DATABASE is None:
            raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
        return self.dest_connection.table(
            name_table, database=self.SNOWFLAKE_DEST_DATABASE
        )

    def list_dest_tables(self) -> List[str]:
        """Names of all tables in the destination Snowflake database."""
        if self.SNOWFLAKE_DEST_DATABASE is None:
            raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
        return self.dest_connection.list_tables(database=self.SNOWFLAKE_DEST_DATABASE)

    def _get_output_table_name(self, table):
        if table.has_name:
            name_table = table.get_name().split(".")[-1]
        else:
            raise ValueError("Must specify name_table!")
        return name_table

    def create_view(self, table, name_table=None, overwrite=False):
        """
        Create a view of a table in the destination Snowflake database.

        Args:
            table (Table): Ibis table object to create a view from.
            name_table (str, optional): Name of the view to create. Defaults to None.
            overwrite (bool, optional): Whether to overwrite the view if it exists. Defaults to False.

        Returns:
            View: Ibis view object created in the destination Snowflake database.
        """
        if self.SNOWFLAKE_DEST_DATABASE is None:
            raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
        name_table = name_table or self._get_output_table_name(table)

        # Check if the destination database exists, if not, create it
        catalog, database = self.SNOWFLAKE_DEST_DATABASE.split(".")
        if not database in self.dest_connection.list_databases(catalog=catalog):
            self.dest_connection.create_database(name=database, catalog=catalog)

        return self.dest_connection.create_view(
            name=name_table.upper(),
            database=self.SNOWFLAKE_DEST_DATABASE,
            obj=table,
            overwrite=overwrite,
            schema=table.schema(),
        )

    def create_table(self, table, name_table=None, overwrite=False, comment=None):
        """
        Materialize a table in the destination Snowflake database.

        Args:
            table: Ibis table object to materialize.
            name_table (str, optional): Name of the table to create. Defaults to None.
            overwrite: Whether to overwrite the table if it exists. Defaults to False.
            comment: Add a comment about the table.

        Returns:
            Table: Ibis table object created in the destination Snowflake database.
        """
        if self.SNOWFLAKE_DEST_DATABASE is None:
            raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")

        name_table = name_table or self._get_output_table_name(table)

        # Check if the destination database exists, if not, create it
        catalog, database = self.SNOWFLAKE_DEST_DATABASE.split(".")
        if not database in self.dest_connection.list_databases(catalog=catalog):
            self.dest_connection.create_database(name=database, catalog=catalog)

        # Unwrap PhenexTable so ibis sees a proper ir.Expr and performs a
        # server-side CTAS instead of falling back to the memtable/pandas path
        # (which triggers client-side query execution and can produce NULL syntax
        # errors or Arrow schema-mismatch failures on multi-batch Snowflake results).
        from phenex.tables import PhenexTable

        if isinstance(table, PhenexTable):
            table = table.table

        return self.dest_connection.create_table(
            name=name_table.upper(),
            database=self.SNOWFLAKE_DEST_DATABASE,
            obj=table,
            overwrite=overwrite,
            schema=table.schema(),
            comment=comment,
        )

    def drop_table(self, name_table):
        """
        Drop a table from the destination Snowflake database.

        Args:
            name_table (str): Name of the table to drop.

        Returns:
            None
        """
        if self.SNOWFLAKE_DEST_DATABASE is None:
            raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
        return self.dest_connection.drop_table(
            name=name_table, database=self.SNOWFLAKE_DEST_DATABASE
        )

    def drop_view(self, name_table):
        """
        Drop a view from the destination Snowflake database.

        Args:
            name_table (str): Name of the view to drop.

        Returns:
            None
        """
        if self.SNOWFLAKE_DEST_DATABASE is None:
            raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
        return self.dest_connection.drop_view(
            name=name_table, database=self.SNOWFLAKE_DEST_DATABASE
        )

connect_dest()

Establishes and returns an Ibis backend connection to the destination Snowflake database.

Returns:

Name	Type	Description
BaseBackend	BaseBackend	Ibis backend connection to the destination Snowflake database.

Source code in phenex/ibis_connect.py

def connect_dest(self) -> BaseBackend:
    """
    Establishes and returns an Ibis backend connection to the destination Snowflake database.

    Returns:
        BaseBackend: Ibis backend connection to the destination Snowflake database.
    """
    return self._connect(
        database=self.SNOWFLAKE_DEST_DATABASE,
    )

connect_source()

Establishes and returns an Ibis backend connection to the source Snowflake database.

Returns:

Name	Type	Description
BaseBackend	BaseBackend	Ibis backend connection to the source Snowflake database.

Source code in phenex/ibis_connect.py

def connect_source(self) -> BaseBackend:
    """
    Establishes and returns an Ibis backend connection to the source Snowflake database.

    Returns:
        BaseBackend: Ibis backend connection to the source Snowflake database.
    """
    return self._connect(
        database=self.SNOWFLAKE_SOURCE_DATABASE,
    )

create_table(table, name_table=None, overwrite=False, comment=None)

Materialize a table in the destination Snowflake database.

Parameters:

Name	Type	Description	Default
table		Ibis table object to materialize.	required
name_table	str	Name of the table to create. Defaults to None.	None
overwrite		Whether to overwrite the table if it exists. Defaults to False.	False
comment		Add a comment about the table.	None

Returns:

Name	Type	Description
Table		Ibis table object created in the destination Snowflake database.

Source code in phenex/ibis_connect.py

def create_table(self, table, name_table=None, overwrite=False, comment=None):
    """
    Materialize a table in the destination Snowflake database.

    Args:
        table: Ibis table object to materialize.
        name_table (str, optional): Name of the table to create. Defaults to None.
        overwrite: Whether to overwrite the table if it exists. Defaults to False.
        comment: Add a comment about the table.

    Returns:
        Table: Ibis table object created in the destination Snowflake database.
    """
    if self.SNOWFLAKE_DEST_DATABASE is None:
        raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")

    name_table = name_table or self._get_output_table_name(table)

    # Check if the destination database exists, if not, create it
    catalog, database = self.SNOWFLAKE_DEST_DATABASE.split(".")
    if not database in self.dest_connection.list_databases(catalog=catalog):
        self.dest_connection.create_database(name=database, catalog=catalog)

    # Unwrap PhenexTable so ibis sees a proper ir.Expr and performs a
    # server-side CTAS instead of falling back to the memtable/pandas path
    # (which triggers client-side query execution and can produce NULL syntax
    # errors or Arrow schema-mismatch failures on multi-batch Snowflake results).
    from phenex.tables import PhenexTable

    if isinstance(table, PhenexTable):
        table = table.table

    return self.dest_connection.create_table(
        name=name_table.upper(),
        database=self.SNOWFLAKE_DEST_DATABASE,
        obj=table,
        overwrite=overwrite,
        schema=table.schema(),
        comment=comment,
    )

create_view(table, name_table=None, overwrite=False)

Create a view of a table in the destination Snowflake database.

Parameters:

Name	Type	Description	Default
table	Table	Ibis table object to create a view from.	required
name_table	str	Name of the view to create. Defaults to None.	None
overwrite	bool	Whether to overwrite the view if it exists. Defaults to False.	False

Returns:

Name	Type	Description
View		Ibis view object created in the destination Snowflake database.

Source code in phenex/ibis_connect.py

def create_view(self, table, name_table=None, overwrite=False):
    """
    Create a view of a table in the destination Snowflake database.

    Args:
        table (Table): Ibis table object to create a view from.
        name_table (str, optional): Name of the view to create. Defaults to None.
        overwrite (bool, optional): Whether to overwrite the view if it exists. Defaults to False.

    Returns:
        View: Ibis view object created in the destination Snowflake database.
    """
    if self.SNOWFLAKE_DEST_DATABASE is None:
        raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
    name_table = name_table or self._get_output_table_name(table)

    # Check if the destination database exists, if not, create it
    catalog, database = self.SNOWFLAKE_DEST_DATABASE.split(".")
    if not database in self.dest_connection.list_databases(catalog=catalog):
        self.dest_connection.create_database(name=database, catalog=catalog)

    return self.dest_connection.create_view(
        name=name_table.upper(),
        database=self.SNOWFLAKE_DEST_DATABASE,
        obj=table,
        overwrite=overwrite,
        schema=table.schema(),
    )

drop_table(name_table)

Drop a table from the destination Snowflake database.

Parameters:

Name	Type	Description	Default
name_table	str	Name of the table to drop.	required

Returns:

Type	Description
	None

Source code in phenex/ibis_connect.py

def drop_table(self, name_table):
    """
    Drop a table from the destination Snowflake database.

    Args:
        name_table (str): Name of the table to drop.

    Returns:
        None
    """
    if self.SNOWFLAKE_DEST_DATABASE is None:
        raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
    return self.dest_connection.drop_table(
        name=name_table, database=self.SNOWFLAKE_DEST_DATABASE
    )

drop_view(name_table)

Drop a view from the destination Snowflake database.

Parameters:

Name	Type	Description	Default
name_table	str	Name of the view to drop.	required

Returns:

Type	Description
	None

Source code in phenex/ibis_connect.py

def drop_view(self, name_table):
    """
    Drop a view from the destination Snowflake database.

    Args:
        name_table (str): Name of the view to drop.

    Returns:
        None
    """
    if self.SNOWFLAKE_DEST_DATABASE is None:
        raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
    return self.dest_connection.drop_view(
        name=name_table, database=self.SNOWFLAKE_DEST_DATABASE
    )

get_dest_table(name_table)

Retrieves a table from the destination Snowflake database.

Parameters:

Name	Type	Description	Default
name_table	str	Name of the table to retrieve.	required

Returns:

Name	Type	Description
Table		Ibis table object from the destination Snowflake database.

Source code in phenex/ibis_connect.py

def get_dest_table(self, name_table):
    """
    Retrieves a table from the destination Snowflake database.

    Args:
        name_table (str): Name of the table to retrieve.

    Returns:
        Table: Ibis table object from the destination Snowflake database.
    """
    if self.SNOWFLAKE_DEST_DATABASE is None:
        raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
    return self.dest_connection.table(
        name_table, database=self.SNOWFLAKE_DEST_DATABASE
    )

get_source_table(name_table)

Retrieves a table from the source Snowflake database.

Parameters:

Name	Type	Description	Default
name_table	str	Name of the table to retrieve.	required

Returns:

Name	Type	Description
Table		Ibis table object from the source Snowflake database.

Source code in phenex/ibis_connect.py

def get_source_table(self, name_table):
    """
    Retrieves a table from the source Snowflake database.

    Args:
        name_table (str): Name of the table to retrieve.

    Returns:
        Table: Ibis table object from the source Snowflake database.
    """
    return self.source_connection.table(
        name_table, database=self.SNOWFLAKE_SOURCE_DATABASE
    )

list_dest_tables()

Names of all tables in the destination Snowflake database.

Source code in phenex/ibis_connect.py

def list_dest_tables(self) -> List[str]:
    """Names of all tables in the destination Snowflake database."""
    if self.SNOWFLAKE_DEST_DATABASE is None:
        raise ValueError("Must specify SNOWFLAKE_DEST_DATABASE!")
    return self.dest_connection.list_tables(database=self.SNOWFLAKE_DEST_DATABASE)