sqlalchemy-diff

from sqlalchemy import create_engine, Column, Integer, String, MetaData, Table from sqlalchemydiff.comparer import Comparer # Define schema for database one (source) metadata_one = MetaData() Table( 'users', metadata_one, Column('id', Integer, primary_key=True), Column('name', String(50)), Column('email', String(100), unique=True) ) # Define schema for database two (target) with a difference metadata_two = MetaData() Table( 'users', metadata_two, Column('id', Integer, primary_key=True), Column('name', String(50)), Column('address', String(200)) # Changed column ) # Create in-memory SQLite engines for demonstration engine_one = create_engine('sqlite:///:memory:') engine_two = create_engine('sqlite:///:memory:') # Create tables in memory metadata_one.create_all(engine_one) metadata_two.create_all(engine_two) # Compare the schemas comparer = Comparer(engine_one, engine_two) result = comparer.compare(one_alias='Source DB', two_alias='Target DB') if result.is_match: print('Schemas are identical!') else: print('Schemas differ! Differences:') for error in result.errors: print(f" - {error.description} (Table: {error.table_name}, Column: {error.column_name or 'N/A'}) [Severity: {error.severity}]") # Example using from_params for automatic engine disposal # (Requires actual database URIs to connect) # try: # comparer_from_params = Comparer.from_params( # 'sqlite:///:memory:', # 'sqlite:///:memory:', # dispose_engines=True # ) # result_from_params = comparer_from_params.compare() # print('\nComparison using from_params:') # if result_from_params.is_match: # print('Schemas are identical (from_params)!') # else: # print('Schemas differ (from_params)!') # for error in result_from_params.errors: # print(f" - {error.description}") # except Exception as e: # print(f"Error with from_params example: {e}")