Schemachange: A Database Change Management tool for Snowflake
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
- breaking Checksum drift for scripts ending with a semicolon when upgrading from v4.3.0/v4.3.1 to v4.3.2. A regression in v4.3.0 accidentally removed trailing semicolon stripping, causing checksum mismatches.
- breaking Minimum `snowflake-connector-python` version was bumped to `>=3.0.0` in v4.3.0, dropping support for `snowflake-connector-python` 2.x.
- gotcha Schemachange, by default, will not create the change history table and will fail if it doesn't exist. It also won't create the database for this table.
- gotcha New CLI authentication parameters (`--snowflake-authenticator`, `--snowflake-private-key-file`, `--snowflake-token-file-path`) were added in v4.1.0, but `--snowflake-private-key-file-pwd` (passphrase) is *intentionally not supported* via CLI for security reasons.
- gotcha Out-of-Order Execution, introduced in v4.3.0, changes how versioned scripts are applied. If enabled, scripts can be applied even if their version is older than the maximum applied version.
- deprecated The `--verbose` flag for command-line output was deprecated.
- deprecated Parameters `private_key_path` and `private_key_passphrase` are slated for removal in the upcoming v5.0.0 major release.
Install
-
pip install schemachange
Imports
- schemachange CLI
import subprocess import os # Most common usage is via CLI subprocess.run(["schemachange", "deploy", "-f", "./migrations"], env=os.environ)
Quickstart
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?")