schemachange

import os import subprocess # Create a dummy migrations folder and script for demonstration if not os.path.exists('migrations'): os.makedirs('migrations') with open('migrations/V1.0.0__initial_setup.sql', 'w') as f: f.write("CREATE SCHEMA IF NOT EXISTS MY_APP;\n") f.write("CREATE TABLE IF NOT EXISTS MY_APP.CUSTOMERS (ID INTEGER, NAME VARCHAR(100));") # Set environment variables for Snowflake connection (replace with your actual details) # For production, use secure methods like secrets management or connections.toml # NOTE: For quickstart, ensure these are actual values or use a dummy Snowflake connection if possible env = os.environ.copy() env['SNOWFLAKE_ACCOUNT'] = os.environ.get('SNOWFLAKE_ACCOUNT', 'your_account_identifier') env['SNOWFLAKE_USER'] = os.environ.get('SNOWFLAKE_USER', 'your_user') env['SNOWFLAKE_PASSWORD'] = os.environ.get('SNOWFLAKE_PASSWORD', 'your_password') env['SNOWFLAKE_ROLE'] = os.environ.get('SNOWFLAKE_ROLE', 'SYSADMIN') env['SNOWFLAKE_WAREHOUSE'] = os.environ.get('SNOWFLAKE_WAREHOUSE', 'COMPUTE_WH') env['SNOWFLAKE_DATABASE'] = os.environ.get('SNOWFLAKE_DATABASE', 'DEMO_DB') env['SCHEMACHANGE_CHANGE_HISTORY_TABLE'] = os.environ.get('SCHEMACHANGE_CHANGE_HISTORY_TABLE', 'DEMO_DB.SCHEMACHANGE.CHANGE_HISTORY') print("Running schemachange deploy...") try: result = subprocess.run( [ "schemachange", "deploy", "-f", "./migrations", "--create-change-history-table", "--verbose" ], env=env, # Pass environment variables to the subprocess check=True, # Raise an exception for non-zero exit codes capture_output=True, # Capture stdout and stderr text=True # Decode stdout/stderr as text ) print("Schemachange deployment successful.") print("STDOUT:", result.stdout) if result.stderr: print("STDERR:", result.stderr) except subprocess.CalledProcessError as e: print(f"Schemachange deployment failed with error: {e}") print("STDOUT:", e.stdout) print("STDERR:", e.stderr) except FileNotFoundError: print("Error: 'schemachange' command not found. Is schemachange installed and in your PATH?")