Copy shared Snowflake Database table by table

snowflake_copy_database_by_table.py

# Useful when you want to make a full copy of a Snowflake Database that is shared with your account
# into a database in your account, but you can't use clone database / etc.
#!/usr/bin/env python
import os
import re
# https://docs.snowflake.com/en/user-guide/python-connector.html
import snowflake.connector


# Connect to snowflake
ctx = snowflake.connector.connect(
    user=os.environ['SNOWFLAKE_USER'],
    password=os.environ['SNOWFLAKE_PASSWORD'],
    account=os.environ['SNOWFLAKE_ACCOUNT']
    )
cs = ctx.cursor()
target_db =  "<PutTargetDBNameHere>"
source_db = "<PutSourceDBNameHere>"
source_schema = "<SourceSchemaHere>"
warehouse = "<PutWareHouseHere>"
bad_chars = '()'
try:
    # Connect to the warehouse and get a list of views
    cs.execute(f"use warehouse {warehouse}")
    cs.execute(f"use database {source_db}")
    cs.execute("select table_name from information_schema.tables where table_type = 'VIEW'")
    views = cs.fetchall()
    # Swap to the target database
    cs.execute(f"use database {target_db}")
    for result in views:
      # Convert the tuple to a string, strip the garbage off.
      table_name = re.sub(f'[{bad_chars}]', '', ''.join(result))
      # Copy the source table to the destination database
      sql_statement = f"create table {table_name} as select * from {source_db}.{source_schema}.{table_name}"
      print(sql_statement)
      try:
        cs.execute(sql_statement)
      except:
        # Tables will error if they exist or are reserved view names like TABLE_CONSTRAINTS, LOAD_HISTORY, etc
        print(f'Error creating {table_name}')

finally:
    cs.close()
ctx.close()