CopyDataFromSQLServerToPostgreSQL/TransferSQLToPostGreSQLMain.py at main · luozhijian/CopyDataFromSQLServerToPostgreSQL · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
# -*- coding: utf-8 -*-
"""
Created on Sun Jan 11 18:56:26 2026

@author: Yanyan
"""

import logging
import os

# ------------------ Example Usage ------------------
from SQLServerToPostgres import SQLServerToPostgres
logger = None

def initialized_log():
    global logger
    logger = logging.getLogger()

    # Avoid adding handlers multiple times (e.g. if initialized_log is called again)
    if logger.handlers:
        logger.info("Logging already initialized.")
        return

    logger.setLevel(logging.DEBUG)  # Capture all levels

    # Create logs directory if it doesn't exist
    os.makedirs("logs", exist_ok=True)

    # Console handler (INFO+)
    ch = logging.StreamHandler()
    ch.setLevel(logging.DEBUG)
    ch_formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(name)s: %(message)s')
    ch.setFormatter(ch_formatter)

    # File handler (DEBUG+)
    fh = logging.FileHandler("logs/transfer_export_only.log")
    fh.setLevel(logging.DEBUG)
    fh_formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(name)s: %(message)s')
    fh.setFormatter(fh_formatter)

    logger.addHandler(ch)
    logger.addHandler(fh)

    logger.propagate = True

    logger.info("SQLServerToPostgres initialized.")


def main():
    initialized_log()
    SQLSERVER_URL = (
        "mssql+pyodbc://@localhost/YourDatabaseName"
        "?driver=ODBC+Driver+17+for+SQL+Server"
        "&trusted_connection=yes"
        "&TrustServerCertificate=yes"
    )
    POSTGRES_URL = (
        "postgresql+psycopg2://postgres:YourPassword@10.0.0.62:5432/YourDatabaseName"
    )

    migrator = SQLServerToPostgres(SQLSERVER_URL, POSTGRES_URL, out_dir=r"F:/Transfer/temp")
    migrator.psql_path = r'D:/Program Files/PostgreSQL/18/bin/psql.exe'

    migrator.test_connections()

    menu = """
    1. Create all tables in target
    2. Copy all tables using SQL directly (SQLAlchemy)
    3. Copy data using BCP and then PSQL
    4. BCP output data only
    5. After data copied, add all indexes
    6. After data copied, similiar to 5, but generate one SQL File for add all indexes and foreign key
    7. Check both tables have same row count
    8. Drop all tables (if you like to redo the process)
    0. Exit
    """
    while True:
        try :
            print(menu)
            choice = input("Enter option (0-6): ").strip()
            if choice == "0":
                logger.info("Exiting.")
                break
            if choice == "1":
                migrator.created_all_tables_and_primary_key()
            elif choice == "2":
                migrator.copy_all_tables_data_directly()
            elif choice == "3":
                migrator.copy_all_tables_data_bcp_psql()
            elif choice == "4":
                migrator.bcp_export_all_tables_only()
            elif choice == "5":
                migrator.add_all_indexes_and_foreign_key( )
            elif choice == "6":
                migrator.generate_pg_indexes_and_fks( 'F:/Transfer/temp/index_fks.sql', {'dbo':'public'} )
            elif choice == "7":
                migrator.check_table_row_counts()
            elif choice == "8":
                migrator.drop_all_tables_in_target()
            else:
                logger.warning("Invalid option. Enter 0-8.")
        except Exception as ex:
            logger.exception(ex)

if __name__ == "__main__":
    try :
        main()
    except Exception as ex :
        logger.error(ex)
    logger.info('SQLServerToPostgres completed')

    logger =None