Snowflake Connector Documentation

import snowflake.connector
from snowflake.connector.cursor import SnowflakeCursor

# Connect using private key
def build_snowflake_cursor() -> SnowflakeCursor:
    conn = snowflake.connector.connect(
        user=SNOWFLAKE_USER,
        account=ACCOUNT_NAME,
        private_key=SNOWFLAKE_PRIVATE_KEY,
        warehouse=SNOWFLAKE_WAREHOUSE,
        database=SNOWFLAKE_DATABASE,
        schema=SCHEMA,
        role=SNOWFLAKE_ROLE,
    )
    return conn.cursor()

import snowflake.connector
from snowflake.connector.cursor import SnowflakeCursor

# Connect using password
def build_snowflake_cursor() -> SnowflakeCursor:
    conn = snowflake.connector.connect(
        user=SNOWFLAKE_USER,
        account=ACCOUNT_NAME,
        password=PASSWORD,
        warehouse=SNOWFLAKE_WAREHOUSE,
        database=SNOWFLAKE_DATABASE,
        schema=SCHEMA,
        role=SNOWFLAKE_ROLE,
    )
    return conn.cursor()

def sample_query(cursor: SnowflakeCursor) -> list[int]:
    query = "select 1"
    return cursor.execute(query).fetchall()

CREATE STREAM example_db.example_schema.example_stream ON VIEW example_db.example_schema.example_table;

from datetime import UTC, datetime
from snowflake.connector import SnowflakeConnection, connect


class SnowflakeStreamExporter:
    def __init__(
        self, snowflake: SnowflakeConnection, database: str, schema: str, stream: str, stage: str
    ):
        self.snowflake = snowflake
        self.database = database
        self.schema = schema
        self.stream = stream
        self.stage = stage

    def export_to_stage(self):
        cur_time = datetime.now(UTC).strftime("%Y%m%d_%H_%M_%S")
        query = f"""
            COPY INTO {self.stage}/{self.schema}/{self.stream}_{cur_time}
            FROM (select object_construct(*) as data from {self.database}.{self.schema}.{self.stream})
            FILE_FORMAT = (TYPE = JSON COMPRESSION = 'GZIP');
        """

        try:
            cursor = self.snowflake.cursor()
            results = cursor.execute(query).fetchall()
            for result in results:
                print(result)
        except Exception as e:
            print(f"Error during COPY INTO operation: {e}")
        finally:
            cursor.close()


if __name__ == "__main__":

    database = "db_name"
    schema = "schema_name"
    stream = "stream_name"
    stage = "@db_name.schema_name.stage_name"

    conn = connect(
        user="username",
        account="account_name",
        database=database,
        schema=schema,
        role="role_name",
        warehouse="warehouse_name",

        # Authenticate using password OR private_key. You only need to use one, not both.
        password="password",
        private_key="private_key",
    )
    exporter = SnowflakeStreamExporter(conn, database, schema, stream, stage)
    exporter.export_to_stage()

copy into @STAGE_NAME/destination/path/
    from (
    select
        block_number,
        timestamp,
        datetime,
        block_hash,
        block_parent_hash,
        nonce,
        sha3_uncles,
        logs_bloom,
        transactions_root,
        state_root,
        receipts_root,
        miner,
        mix_hash,
        extra_data,
        difficulty,
        total_difficulty,
        size,
        gas_limit,
        gas_used,
        transaction_count,
        base_fee_per_gas,
        is_reorg,
        date_created,
        date_updated
    from ethereum.blocks
    where datetime >= '2024-06-01' and datetime < '2024-07-01'
)
partition by (
    'year=' ||
    to_char(date(date_trunc(year, datetime)), 'YYYY') || '/' ||
    'month=' ||
    to_char(date(date_trunc(month, datetime)), 'MM') || '/' ||
    'day=' ||
    lpad(to_char(extract(day from datetime)), 2, '0') || '/' ||
    'hour=' ||
    lpad(to_char(extract(hour from datetime)), 2, '0')
)
file_format = (type = parquet)
header = true
max_file_size = 5242880000
detailed_output = true
;

import os
        sf_cur.execute(table_sql)
        for row in sf_cur.fetchall():
            table_list.append(row[0].lower())

    return table_list


def write_local_file(sf_conn, schema: str, table_name: str, chunksize: int = 100000, output_dir: str = "output"):
    row_count = 0
    tgt_table_name = f"{schema.lower()}.{table_name.lower()}"
    os.makedirs(os.path.join(output_dir, table_name), exist_ok=True)

    read_sql = jinjafy_sql(READ_TEMPLATE, {"tgt_table": tgt_table_name})
    print(read_sql.strip())

    chunk_count = 0
    for df in pd.read_sql(read_sql, sf_conn, chunksize=chunksize):
        file_name = f"{output_dir}/{table_name}/{table_name}_{chunk_count}.csv"
        row_count += len(df)
        df.columns = map(str.lower, df.columns)
        print(f"Loading chunk {row_count:,} rows")
        df.to_csv(file_name, index=False)
        chunk_count += 1

    return row_count


class SnowflakeBackfillExporter:
    def __init__(
        self,
        snowflake: snowflake.connector.SnowflakeConnection,
        chunksize: int = 100000,
        output_dir: str = "output",
    ):
        self.snowflake = snowflake
        self.database = os.getenv("SNOWFLAKE_DATABASE")
        self.schema = os.getenv("SNOWFLAKE_SCHEMA")
        self.chunksize = chunksize
        self.output_dir = output_dir

    def run_backfill(self):
        sf_conn = get_snowflake_connection()

        row_count = table_count = 0

        # Get list of tables in Snowflake to export
        table_list = get_snowflake_tables_in_schema(sf_conn, self.schema)
        print(f"Tables in {self.schema}: {table_list}")

        for table_name in table_list:
            table_count += 1
            table_row_count = 0

            # Query and write data to local file
            # Alter this to export to desired destination
            table_row_count = write_local_file(
                sf_conn, self.schema, table_name=table_name, chunksize=self.chunksize, output_dir=self.output_dir
            )

            row_count += table_row_count
            print(f"Exported {table_row_count:,} rows from {table_name}\n")

        print(f"Exported {row_count:,} rows from {table_count} tables")

        sf_conn.close()


if __name__ == "__main__":
    sf_conn = get_snowflake_connection()
    exporter = SnowflakeBackfillExporter(sf_conn)
    exporter.run_backfill()