Table Comparison

import cx_Oracle

import csv

def get_schema_info(connection):

    cursor = connection.cursor()

    query = """

    SELECT owner, table_name, column_name, data_type, data_length

    FROM all_tab_columns

    ORDER BY owner, table_name, column_id

    """

    cursor.execute(query)

    schema_info = {}

    for owner, table_name, column_name, data_type, data_length in cursor:

        full_table_name = f"{owner}.{table_name}"

        if full_table_name not in schema_info:

            schema_info[full_table_name] = []

        schema_info[full_table_name].append((column_name, data_type, data_length))

    cursor.close()

    return schema_info

def compare_schemas(schema_a, schema_b):

    only_in_a = set(schema_a.keys()) - set(schema_b.keys())

    only_in_b = set(schema_b.keys()) - set(schema_a.keys())

    in_both = set(schema_a.keys()) & set(schema_b.keys())

    differences = {}

    for table in in_both:

        columns_a = set(schema_a[table])

        columns_b = set(schema_b[table])

        if columns_a != columns_b:

            differences[table] = {'A': columns_a - columns_b, 'B': columns_b - columns_a}

    return only_in_a, only_in_b, differences

def save_to_csv(filename, only_in_a, only_in_b, differences):

    with open(filename, 'w', newline='') as file:

        writer = csv.writer(file)

        writer.writerow(["Full Table Name", "Only in DB A", "Only in DB B", "Differences in both DBs"])

        for table in sorted(only_in_a):

            writer.writerow([table, "Yes", "No", ""])

        for table in sorted(only_in_b):

            writer.writerow([table, "No", "Yes", ""])

        for table, diff in sorted(differences.items()):

            diff_a = "; ".join([f"{col[0]} ({col[1]}, {col[2]})" for col in diff['A']])

            diff_b = "; ".join([f"{col[0]} ({col[1]}, {col[2]})" for col in diff['B']])

            writer.writerow([table, "", "", f"A: {diff_a} | B: {diff_b}"])

# Database A connection details

dsn_a = cx_Oracle.makedsn("host_A", port_A, service_name="service_name_A")

conn_a = cx_Oracle.connect(user="username_A", password="password_A", dsn=dsn_a)

# Database B connection details

dsn_b = cx_Oracle.makedsn("host_B", port_B, service_name="service_name_B")

conn_b = cx_Oracle.connect(user="username_B", password="password_B", dsn=dsn_b)

try:

    schema_a = get_schema_info(conn_a)

    schema_b = get_schema_info(conn_b)

    only_in_a, only_in_b, differences = compare_schemas(schema_a, schema_b)

    save_to_csv('database_comparison.csv', only_in_a, only_in_b, differences)

finally:

    conn_a.close()

    conn_b.close()

print("Comparison saved to database_comparison.csv")

index Comparison

import cx_Oracle

import csv

def get_index_info(connection):

    cursor = connection.cursor()

    query = """

    SELECT idx.table_owner || '.' || idx.table_name as table_name,

           idx.index_name,

           LISTAGG(icol.column_name, ', ') WITHIN GROUP (ORDER BY icol.column_position) as columns

    FROM all_indexes idx

    JOIN all_ind_columns icol ON idx.index_name = icol.index_name AND idx.table_owner = icol.table_owner

    GROUP BY idx.table_owner, idx.table_name, idx.index_name

    ORDER BY idx.table_owner, idx.table_name, idx.index_name

    """

    cursor.execute(query)

    index_info = {}

    for table_name, index_name, columns in cursor:

        if table_name not in index_info:

            index_info[table_name] = []

        index_info[table_name].append((index_name, columns))

    cursor.close()

    return index_info

def compare_indexes(indexes_a, indexes_b):

    differences = {}

    for table, indexes in indexes_a.items():

        indexes_b_for_table = indexes_b.get(table, [])

        diff = set(indexes) - set(indexes_b_for_table)

        if diff:

            differences[table] = {'A': diff, 'B': set()}

    for table, indexes in indexes_b.items():

        indexes_a_for_table = indexes_a.get(table, [])

        diff = set(indexes) - set(indexes_a_for_table)

        if diff:

            if table in differences:

                differences[table]['B'] = diff

            else:

                differences[table] = {'A': set(), 'B': diff}

    return differences

def save_indexes_to_csv(filename, differences):

    with open(filename, 'w', newline='') as file:

        writer = csv.writer(file)

        writer.writerow(["Table Name", "Indexes Only in DB A", "Indexes Only in DB B"])

        for table, diff in sorted(differences.items()):

            diff_a = "; ".join([f"{idx[0]} ({idx[1]})" for idx in diff['A']])

            diff_b = "; ".join([f"{idx[0]} ({idx[1]})" for idx in diff['B']])

            writer.writerow([table, diff_a, diff_b])

# Database A connection details

dsn_a = cx_Oracle.makedsn("host_A", port_A, service_name="service_name_A")

conn_a = cx_Oracle.connect(user="username_A", password="password_A", dsn=dsn_a)

# Database B connection details

dsn_b = cx_Oracle.makedsn("host_B", port_B, service_name="service_name_B")

conn_b = cx_Oracle.connect(user="username_B", password="password_B", dsn=dsn_b)

try:

    indexes_a = get_index_info(conn_a)

    indexes_b = get_index_info(conn_b)

    index_differences = compare_indexes(indexes_a, indexes_b)

    save_indexes_to_csv('database_index_comparison.csv', index_differences)

finally:

    conn_a.close()

    conn_b.close()

print("Index comparison saved to database_index_comparison.csv")

PL/SQL Comparison

import cx_Oracle

import csv

def get_plsql_info(connection):

    cursor = connection.cursor()

    query = """

    SELECT name, type, line, text

    FROM all_source

    WHERE type IN ('PROCEDURE', 'FUNCTION')

    ORDER BY name, type, line

    """

    cursor.execute(query)

    plsql_info = {}

    for name, type, line, text in cursor:

        key = f"{type}:{name}"

        if key not in plsql_info:

            plsql_info[key] = []

        plsql_info[key].append(text.strip())

    cursor.close()

    return plsql_info

def compare_plsql(plsql_a, plsql_b):

    only_in_a = set(plsql_a.keys()) - set(plsql_b.keys())

    only_in_b = set(plsql_b.keys()) - set(plsql_a.keys())

    in_both = set(plsql_a.keys()) & set(plsql_b.keys())

    differences = {}

    for key in in_both:

        if "".join(plsql_a[key]) != "".join(plsql_b[key]):

            differences[key] = {'A': plsql_a[key], 'B': plsql_b[key]}

    return only_in_a, only_in_b, differences

# Rest of the script (database connections, schema/index/privilege comparisons)

# ...

# PL/SQL Comparison

plsql_a = get_plsql_info(conn_a)

plsql_b = get_plsql_info(conn_b)

plsql_differences = compare_plsql(plsql_a, plsql_b)

# Output to CSV

# Due to the potential length of source code, consider how best to format this in CSV

# ...

# Close connections

# ...

Privledge Comparison

import cx_Oracle

import csv

def get_privileges_info(connection):

    cursor = connection.cursor()

    query = """

    SELECT grantee, table_name, privilege

    FROM dba_tab_privs

    WHERE owner NOT IN ('SYS', 'SYSTEM')

    ORDER BY grantee, table_name, privilege

    """

    cursor.execute(query)

    privileges_info = {}

    for grantee, table_name, privilege in cursor:

        key = (grantee, table_name)

        if key not in privileges_info:

            privileges_info[key] = []

        privileges_info[key].append(privilege)

    cursor.close()

    return privileges_info

def compare_privileges(privs_a, privs_b):

    only_in_a = set(privs_a.keys()) - set(privs_b.keys())

    only_in_b = set(privs_b.keys()) - set(privs_a.keys())

    common = set(privs_a.keys()) & set(privs_b.keys())

    differences = {}

    for key in common:

        if set(privs_a[key]) != set(privs_b[key]):

            differences[key] = {'A': privs_a[key], 'B': privs_b[key]}

    return only_in_a, only_in_b, differences

def save_privileges_to_csv(filename, only_in_a, only_in_b, differences):

    with open(filename, 'w', newline='') as file:

        writer = csv.writer(file)

        writer.writerow(["Grantee", "Table Name", "Privileges Only in DB A", "Privileges Only in DB B", "Differences in both DBs"])

        for grantee, table in sorted(only_in_a):

            writer.writerow([grantee, table, ", ".join(privs_a[(grantee, table)]), "", ""])

        for grantee, table in sorted(only_in_b):

            writer.writerow([grantee, table, "", ", ".join(privs_b[(grantee, table)]), ""])

        for (grantee, table), diff in sorted(differences.items()):

            diff_a = ", ".join(diff['A'])

            diff_b = ", ".join(diff['B'])

            writer.writerow([grantee, table, diff_a, diff_b, "Diff"])

# Database A connection details

dsn_a = cx_Oracle.makedsn("host_A", port_A, service_name="service_name_A")

conn_a = cx_Oracle.connect(user="username_A", password="password_A", dsn=dsn_a)

# Database B connection details

dsn_b = cx_Oracle.makedsn("host_B", port_B, service_name="service_name_B")

conn_b = cx_Oracle.connect(user="username_B", password="password_B", dsn=dsn_b)

try:

    privs_a = get_privileges_info(conn_a)

    privs_b = get_privileges_info(conn_b)

    only_in_a, only_in_b, priv_differences = compare_privileges(privs_a, privs_b)

    save_privileges_to_csv('database_privileges_comparison.csv', only_in_a, only_in_b, priv_differences)

finally:

    conn_a.close()

    conn_b.close()

print("Privileges comparison saved to database_privileges_comparison.csv")