Schemachange: A Database Change Management tool for Snowflake

4.3.2 · active · verified Wed Apr 15

Schemachange is an open-source Python-based database change management (DCM) tool designed for Snowflake, inspired by the Flyway database migration tool. It follows an imperative-style approach, allowing users to track and apply SQL and CLI migration scripts in a version-controlled, repeatable, and auditable manner. It integrates seamlessly with CI/CD pipelines to automate database deployments. The current version is 4.3.2, with minor releases typically occurring every 4-6 weeks and patch releases issued as needed for critical bug fixes.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to set up a basic `schemachange` project and run a deployment using Python's `subprocess` module to execute the `schemachange` CLI. It creates a dummy `migrations` folder and an initial SQL script, then attempts to deploy it to a Snowflake instance using credentials provided via environment variables. The `--create-change-history-table` flag is included to automatically create the change history table if it doesn't exist. Ensure your Snowflake account, user, password, role, warehouse, and database environment variables are correctly set for authentication.

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?")

view raw JSON →