Bermigrasi dari PostgreSQL ke Microsoft SQL dengan mssql-python

Banyak tim Python pertama kali mempelajari PostgreSQL. Saat beban kerja Anda membutuhkan fitur seperti tabel temporal, semantik lengkapMERGE, atau indeks penyimpan kolom, migrasikan ke Microsoft SQL. Panduan ini mencakup keputusan utama dan perubahan kode untuk memindahkan aplikasi Python dari PostgreSQL (menggunakan psycopg2 atau psycopg3) ke Microsoft SQL dengan menggunakan mssql-python driver.

Note

Jika Anda bermigrasi dari Azure Database for PostgreSQL, kedua layanan mendukung autentikasi Microsoft Entra dan identitas terkelola. Perubahan kode dalam panduan ini berlaku terlepas dari apakah sumber PostgreSQL Anda dikelola sendiri atau dihosting Azure.

Apa yang Anda peroleh dengan pindah ke Microsoft SQL

Microsoft SQL menyertakan kemampuan yang menyederhanakan keamanan, kepatuhan, dan operasi untuk beban kerja produksi. Pahami fitur-fitur ini sebelum Anda mulai bermigrasi sehingga Anda dapat memanfaatkannya selama transisi:

  • Penyamaran data dinamis dan keamanan tingkat baris. Tutupi kolom untuk pengguna yang tidak memerlukan akses penuh, dan batasi visibilitas baris berdasarkan kebijakan keamanan. Fitur ini dapat digunakan dengan driver apa pun.
  • Tabel temporal (versi sistem). Microsoft SQL melacak riwayat baris secara otomatis. Tidak ada pemicu, tidak ada tabel audit, tidak ada kode aplikasi.
  • Semantik lengkap MERGE. Satu pernyataan mencakup INSERT, UPDATE, dan DELETE dengan klausa OUTPUT untuk rekam audit. Klausa PostgreSQL ON CONFLICT hanya mencakup sisipkan atau pembaruan pada satu batasan.
  • Indeks penyimpanan kolom. Tambahkan penyimpanan kolom ke tabel yang ada untuk beban kerja OLTP/analitik hibrid. Tidak diperlukan database analitik terpisah.
  • Autentikasi Microsoft Entra ID. Terhubung dengan identitas terkelola, perwakilan layanan, atau masuk interaktif. Azure Database for PostgreSQL juga mendukung autentikasi Microsoft Entra, jadi jika Anda sudah menggunakannya, transisinya mudah.

Pasang driver

Sebelum memulai, pastikan Anda memiliki Python 3.10 atau lebih baru dan database SQL target.

Membuat database SQL

Buat atau sambungkan ke database SQL di salah satu platform berikut:

Driver PostgreSQL memerlukan pustaka asli eksternal.

# psycopg2 requires pg_config, libpq-dev, and platform-specific build tools
sudo apt-get install libpq-dev  # Debian/Ubuntu
pip install psycopg2

Driver mssql-python menyertakan lapisan native-nya. Di Windows, Anda tidak memerlukan pengelola driver eksternal atau paket sistem.

pip install mssql-python

Di Linux dan macOS, instal sekumpulan kecil pustaka sistem yang didokumentasikan dalam Penginstalan. Tidak ada yang setara dengan pg_config atau libpq-dev.

Perbarui kode koneksi

Bagian berikut mencakup perubahan utama pada string koneksi, autentikasi, pengelola konteks, dan pengumpulan.

Rangkaian koneksi

psycopg2 menggunakan string DSN atau argumen kata kunci.

import psycopg2

conn = psycopg2.connect(
    host="<server>",
    dbname="<database>",
    user="<username>",
    password="<password>"
)

mssql-python juga mendukung argumen kata kunci, yang menghindari masalah pengkodean URL yang sering dialami string koneksi SQLAlchemy ketika kata sandi berisi @, ;atau {} karakter.

import mssql_python

conn = mssql_python.connect(
    server="<server>.database.windows.net",
    database="<database>",
    authentication="ActiveDirectoryDefault",
    encrypt="yes"
)

Atau gunakan string koneksi.

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes;"
)

Untuk kumpulan lengkap kata kunci string koneksi, lihat Connection strings.

Authentication

Autentikasi PostgreSQL biasanya menggunakan pg_hba.conf aturan dengan nama pengguna dan kata sandi. Azure Database for PostgreSQL juga mendukung autentikasi Microsoft Entra. Microsoft SQL mendukung beberapa mode autentikasi melalui satu kata kunci koneksi:

Pendekatan PostgreSQL padanan mssql-python
Nama pengguna dan kata sandi UID=...;PWD=...;
Enkripsi SSL/TLS Encrypt=yes;(diaktifkan secara default untuk Azure SQL)
Entra auth (Azure PostgreSQL) Authentication=ActiveDirectoryDefault; (tanpa kata sandi)
Identitas terkelola (Azure PostgreSQL) Authentication=ActiveDirectoryMSI;
Service principal (Azure PostgreSQL) Authentication=ActiveDirectoryServicePrincipal;

Gunakan ActiveDirectoryDefault untuk pengembangan lokal. Ini dirantai melalui Azure CLI, variabel lingkungan, dan identitas terkelola secara otomatis. Untuk produksi, gunakan mode tertentu seperti ActiveDirectoryMSI (identitas terkelola) atau ActiveDirectoryServicePrincipal untuk menghindari perjalanan rantai kredensial yang lambat. Lihat autentikasi Microsoft Entra untuk ketujuh mode autentikasi.

Manajer konteks

Kedua driver mendukung pengelola konteks, tetapi perilakunya berbeda:

psycopg2 with conn: melakukan commit saat berhasil dan rollback saat terjadi exception, tetapi tidak menutup koneksi:

with psycopg2.connect(...) as conn:
    with conn.cursor() as cur:
        cur.execute("INSERT INTO ...")
    # conn.commit() happens automatically on success
# Connection is still open here
conn.close()  # Must close explicitly

mssql-python with conn: menutup koneksi saat keluar. Pekerjaan yang tidak berkomitmen dikembalikan:

with mssql_python.connect(...) as conn:
    with conn.cursor() as cursor:
        cursor.execute("INSERT INTO ...")
    conn.commit()
# Connection is closed here

Pemanfaatan koneksi

psycopg2 memerlukan penyiapan dan pengelolaan kumpulan koneksi yang eksplisit.

from psycopg2 import pool

connection_pool = pool.ThreadedConnectionPool(1, 10, dsn="...")
conn = connection_pool.getconn()
# ... use conn ...
connection_pool.putconn(conn)

Driver mssql-python memiliki pooling bawaan yang diaktifkan secara default. Tidak diperlukan penyiapan.

# Pooling is automatic. Each connect() call reuses pooled connections.
conn = mssql_python.connect(...)

Konfigurasikan ukuran kumpulan jika defaultnya tidak sesuai dengan beban kerja Anda.

import mssql_python

mssql_python.pooling(max_size=20, idle_timeout=300)

Untuk panduan tentang ukuran kolam dan pemecahan masalah kelelahan kolam, lihat Pengumpulan koneksi.

Perbedaan dialek SQL

Tabel berikut memetakan pola PostgreSQL umum ke padanan Transact-SQL (T-SQL) mereka:

PostgreSQL SQL Server (T-SQL) Catatan
SERIAL / BIGSERIAL int IDENTITY(1,1) Microsoft SQL menggunakan IDENTITY untuk kenaikan otomatis.
TEXT nvarchar(max) Gunakan nvarchar untuk Unicode. Lebih suka nvarchar(4000) atau lebih pendek jika data memungkinkan.
BOOLEAN bit PostgreSQL menerima true/false; Microsoft SQL menggunakan 1/0.
BYTEA varbinary(max) Konsep yang sama, nama yang berbeda.
JSONB nvarchar(max) dengan fungsi JSON Microsoft SQL menyimpan JSON sebagai teks dan memvalidasi dengan ISJSON(). Lihat data JSON.
TIMESTAMP WITH TIME ZONE datetimeoffset Keduanya menyimpan offset. Lihat Penanganan tanggal dan waktu.
INTERVAL Tidak ada yang setara langsung Hitung dengan DATEADD() dan DATEDIFF().
ARRAY Tidak ada yang setara langsung Gunakan tabel terpisah, array JSON, atau STRING_SPLIT().
UUID uniqueidentifier Driver mssql-python memetakan uuid.UUID secara asli. Lihat Konfigurasi modul.
NOW() / CURRENT_TIMESTAMP GETDATE() atau SYSDATETIME() SYSDATETIME() memberikan presisi yang lebih tinggi.
LIMIT 10 OFFSET 20 OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY Membutuhkan ORDER BY klausa.
\|\| (koncat string) + atau CONCAT() CONCAT() menangani nilai NULL.
COALESCE(a, b) COALESCE(a, b) atau ISNULL(a, b) COALESCE identik di keduanya.
string_agg(col, ',') STRING_AGG(col, ',') Tersedia di SQL Server 2017+.
RETURNING id OUTPUT INSERTED.id Gunakan OUTPUT dalam pernyataan , INSERT, atau UPDATEDELETE.
ON CONFLICT ... DO UPDATE MERGE pernyataan MERGE mendukung INSERT + UPDATE + DELETE dalam satu pernyataan. Lihat Pola penulisan ulang kueri.
EXPLAIN ANALYZE SET STATISTICS IO ON; SET STATISTICS TIME ON; Atau gunakan rencana eksekusi di SSMS / Azure Data Studio.
\d tablename sp_help 'tablename' Atau kueri INFORMATION_SCHEMA.COLUMNS.
pg_dump bcp, BACKUP DATABASE Gunakan bulkcopy() untuk pemuatan data terprogram dari Python.

CREATE TABLE contoh

PostgreSQL:

CREATE TABLE IF NOT EXISTS products (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    price NUMERIC(10, 2) DEFAULT 0.0,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    metadata JSONB,
    is_active BOOLEAN DEFAULT TRUE
);

SQL Server:

IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'products')
CREATE TABLE products (
    id int IDENTITY(1,1) PRIMARY KEY,
    name nvarchar(100) NOT NULL,
    price decimal(10,2) DEFAULT 0.0,
    created_at datetimeoffset DEFAULT SYSDATETIMEOFFSET(),
    metadata nvarchar(max),
    is_active bit DEFAULT 1
);

Pola penulisan ulang kueri

Bagian berikut menunjukkan pola kueri PostgreSQL umum dan padanan T-SQL-nya.

Pagination

PostgreSQL:

cursor.execute("SELECT * FROM products ORDER BY name LIMIT %s OFFSET %s", (10, 20))

mssql-python:

cursor.execute(
    "SELECT * FROM Production.Product ORDER BY Name OFFSET ? ROWS FETCH NEXT ? ROWS ONLY",
    (20, 10)
)

Urutan parameter dibalik. Microsoft SQL menempatkan OFFSET sebelum FETCH NEXT.

Upsert (sisipkan atau perbarui)

PostgreSQL ON CONFLICT menangani insert-or-update pada satu batasan:

cursor.execute("""
    INSERT INTO settings (key, value)
    VALUES (%s, %s)
    ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value
""", (key, value))

Microsoft SQL MERGE menangani INSERT, UPDATE, dan DELETE dalam satu pernyataan. Gunakan USING klausa dengan alias parameter:

cursor.execute("""
    MERGE #Settings AS target
    USING (SELECT ? AS [key], ? AS value) AS source
    ON target.[key] = source.[key]
    WHEN MATCHED THEN UPDATE SET value = source.value
    WHEN NOT MATCHED THEN INSERT ([key], value) VALUES (source.[key], source.value);
""", (key, value))

Untuk upsert dalam jumlah besar, tempatkan baris-baris dalam tabel sementara dengan menggunakan bulkcopy(), lalu lakukan MERGE dari tabel tersebut. Untuk informasi selengkapnya, lihat Peningkatan massal dengan tabel penahapan.

Mendapatkan ID yang dimasukkan

PostgreSQL:

cursor.execute(
    "INSERT INTO products (name) VALUES (%s) RETURNING id",
    ("Widget",)
)
product_id = cursor.fetchone()[0]

mssql-python:

cursor.execute(
    "INSERT INTO #Products (Name) OUTPUT INSERTED.ProductID VALUES (%(name)s)",
    {"name": "Widget"}
)
product_id = cursor.fetchval()

OUTPUT INSERTED bekerja dengan INSERT, UPDATE, dan DELETE pernyataan. Ini dapat mengembalikan beberapa kolom.

Penanda parameter

psycopg2 digunakan %s untuk parameter posisi dan %(name)s untuk parameter bernama. Driver mssql-python menggunakan ? untuk parameter posisional dan %(name)s untuk parameter bernama:

psycopg2:

cursor.execute("SELECT * FROM products WHERE id = %s", (42,))
cursor.execute("SELECT * FROM products WHERE id = %(id)s", {"id": 42})

mssql-python:

cursor.execute("SELECT * FROM Production.Product WHERE ProductID = ?", (42,))
cursor.execute(
    "SELECT * FROM Production.Product WHERE ProductID = %(id)s", {"id": 42}
)

Perbedaan transaksi dan komit otomatis

PostgreSQL (psycopg2) membuka transaksi secara otomatis pada perintah pertama dan memerlukan commit() secara eksplisit:

conn = psycopg2.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()

mssql-python Driver bekerja dengan cara yang sama secara default. Komit otomatis dinonaktifkan, dan Anda memanggil commit() secara eksplisit:

conn = mssql_python.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()

Untuk mengaktifkan autocommit:

psycopg2:

conn = psycopg2.connect(...)
conn.autocommit = True

mssql-python:

conn = mssql_python.connect(..., autocommit=True)
# or: conn.autocommit = True

Lihat bagian Pengelolaan transaksi untuk tingkat isolasi, titik simpan, dan pola percobaan ulang saat deadlock.

Pertimbangan jenis

Bagian berikut mencakup perbedaan pemetaan jenis yang paling umum antara PostgreSQL dan Microsoft SQL.

JSON

PostgreSQL memiliki operator pengindeksan dan kueri asli JSONB (->, ->>, ). @> Microsoft SQL menyimpan JSON sebagai nvarchar(max) dan menyediakan fungsi untuk mengkueri:

PostgreSQL SQL Server
data->>'name' JSON_VALUE(data, '$.name')
data->'items' JSON_QUERY(data, '$.items')
data @> '{"active": true}' JSON_VALUE(data, '$.active') = 'true'
jsonb_array_length(data) (SELECT COUNT(*) FROM OPENJSON(data))

Di Python, kedua pendekatan digunakan json.dumps() untuk serialisasi:

import json

cursor.execute(
    "INSERT INTO #Settings ([key], data) VALUES (%(key)s, %(data)s)",
    {"key": "config", "data": json.dumps({"theme": "dark", "lang": "en"})}
)

Lihat data JSON untuk panduan lengkap tentang penyimpanan JSON dan pola kueri.

UUID (Pengidentifikasi Unik Universal)

Baik PostgreSQL dan mssql-python memetakan uuid.UUID secara asli:

import uuid

cursor.execute(
    "INSERT INTO #Events (EventID, Name) VALUES (%(event_id)s, %(name)s)",
    {"event_id": uuid.uuid4(), "name": "signup"}
)

Lihat Konfigurasi modul untuk native_uuid opsi koneksi.

Tanggalwaktu dan zona waktu

PostgreSQL mengonversi TIMESTAMPTZ ke UTC saat disimpan. Microsoft SQL datetimeoffset mempertahankan offset asli:

from datetime import datetime, timezone, timedelta

eastern = timezone(timedelta(hours=-5))
dt = datetime(2025, 6, 15, 14, 30, tzinfo=eastern)

# PostgreSQL stores as UTC: 2025-06-15 19:30:00+00
# SQL Server stores as-is: 2025-06-15 14:30:00-05:00
cursor.execute("INSERT INTO #Events (EventTime) VALUES (%(event_time)s)", {"event_time": dt})

Jika Anda memerlukan penyimpanan UTC yang konsisten, konversi dalam Python sebelum memasukkan:

dt_utc = dt.astimezone(timezone.utc)
cursor.execute("INSERT INTO #Events (EventTime) VALUES (%(event_time)s)", {"event_time": dt_utc})

Lihat Penanganan tanggal dan waktu untuk pemetaan tipe lengkap.

Array

PostgreSQL mendukung kolom array asli (INTEGER[], TEXT[]). Microsoft SQL tidak memiliki jenis array. Alternatif umum:

  1. Tabel terpisah (dinormalisasi). Terbaik untuk data yang dapat dikueri dan diindeks.
  2. Array JSON disimpan di nvarchar(max). Cocok untuk metadata opak.
  3. String yang dipisahkan koma dengan STRING_SPLIT(). Sederhana tapi terbatas.
# Option 1: Normalized table
cursor.execute("INSERT INTO #ProductTags (ProductID, Tag) VALUES (%(product_id)s, %(tag)s)", {"product_id": 1, "tag": "electronics"})
cursor.execute("INSERT INTO #ProductTags (ProductID, Tag) VALUES (%(product_id)s, %(tag)s)", {"product_id": 1, "tag": "sale"})

# Option 2: JSON array
import json
tags = json.dumps(["electronics", "sale"])
cursor.execute("INSERT INTO #Products (Name, Tags) VALUES (%(name)s, %(tags)s)", {"name": "Widget", "tags": tags})

Ekasandi

PostgreSQL menyimpan semua teks sebagai UTF-8 secara default. Microsoft SQL membedakan antara varchar (pengkodean halaman kode) dan nvarchar (UTF-16). mssql-python Driver mengirimkan nilai Python str sebagai nvarchar secara default, sehingga teks Unicode berfungsi tanpa konfigurasi tambahan. Jika skema Anda menggunakan kolom varchar dan Anda perlu menghindari konversi implisit, gunakan setinputsizes() untuk menentukan jenis kolom. Lihat Data string dan Unicode untuk detail pengodean.

Pemuatan massal dan pemindahan data

PostgreSQL digunakan COPY untuk operasi massal. mssql-python menyediakan bulkcopy():

psycopg2:

with open("data.csv") as f:
    cursor.copy_expert("COPY products FROM STDIN CSV HEADER", f)

mssql-python:

import csv

with open("data.csv", newline="") as f:
    reader = csv.reader(f)
    next(reader)  # Skip header
    rows = [tuple(row) for row in reader]

cursor.bulkcopy("##Products", rows)

Untuk file besar, gunakan generator untuk menghindari pemuatan seluruh file ke dalam memori:

import csv

def csv_rows(path):
    with open(path, newline="") as f:
        reader = csv.reader(f)
        next(reader)  # Skip header
        for row in reader:
            yield tuple(row)

cursor.bulkcopy("##Products", csv_rows("data.csv"), batch_size=5000)

Lihat Operasi penyalinan massal untuk pemetaan kolom, penanganan identitas, dan tips performa.

Skema dan migrasi data

Gunakan pendekatan ini untuk memigrasikan database PostgreSQL yang ada:

  1. Ekspor skema. Gunakan pg_dump --schema-only untuk mendapatkan DDL. Untuk detail opsi dan kasus tepi (kepemilikan, hak istimewa, ekstensi, dan pemfilteran), lihat referensi PostgreSQLpg_dump. Tulis ulang DDL menggunakan tabel perbedaan dialek SQL .
  2. Buat tabel di Microsoft SQL. Jalankan DDL yang ditulis ulang terhadap database target Anda.
  3. Ekspor data. Gunakan pg_dump --data-only --format=csv atau kueri setiap tabel dengan psycopg2. Untuk himpunan data besar dan sakelar kompatibilitas, tinjau dokumentasi PostgreSQLpg_dump, terutama bagian opsi.
  4. Muat data dengan salinan massal. Baca urutan kolom tujuan dari katalog agar Anda tidak perlu menetapkan daftar kolom secara hardcode untuk setiap tabel, lalu kirimkan data setiap tabel secara streaming ke Microsoft SQL. Berikut adalah contoh skrip:
import json
import psycopg2
from psycopg2 import sql
import mssql_python

pg_conn = psycopg2.connect(host="<pgserver>", dbname="<database>", user="<username>", password="<password>")
sql_conn = mssql_python.connect(
    server="<server>.database.windows.net",
    database="<database>",
    authentication="ActiveDirectoryDefault",
    encrypt="yes"
)

def table_columns(cursor, table):
    """Return the ordered column names and identity column from the catalog."""
    cursor.execute(
        "SELECT c.name, c.is_identity FROM sys.columns AS c "
        "WHERE c.object_id = OBJECT_ID(?) ORDER BY c.column_id",
        (table,)
    )
    columns, identity = [], None
    for name, is_identity in cursor.fetchall():
        columns.append(name)
        if is_identity:
            identity = name
    return columns, identity

def parse_pg_table_name(qualified_name):
    """Split a PostgreSQL table name into schema and table parts."""
    if "." in qualified_name:
        schema_name, table_name = qualified_name.split(".", 1)
    else:
        schema_name, table_name = "public", qualified_name
    return schema_name, table_name

def parse_sql_table_name(qualified_name):
    """Split a SQL Server table name into schema and table parts."""
    if "." in qualified_name:
        schema_name, table_name = qualified_name.split(".", 1)
    else:
        schema_name, table_name = "dbo", qualified_name
    return schema_name, table_name

def dependency_order(pg_cursor, table_names, schema_name="public"):
    """Topologically sort tables by foreign key dependencies."""
    table_set = set(table_names)
    incoming = {name: 0 for name in table_set}
    edges = {name: set() for name in table_set}

    pg_cursor.execute(
        """
        SELECT
            child.relname AS child_table,
            parent.relname AS parent_table
        FROM pg_constraint c
        JOIN pg_class child ON c.conrelid = child.oid
        JOIN pg_namespace child_ns ON child.relnamespace = child_ns.oid
        JOIN pg_class parent ON c.confrelid = parent.oid
        JOIN pg_namespace parent_ns ON parent.relnamespace = parent_ns.oid
        WHERE c.contype = 'f'
          AND child_ns.nspname = %s
          AND parent_ns.nspname = %s
        """,
        (schema_name, schema_name),
    )

    for child, parent in pg_cursor.fetchall():
        if child in table_set and parent in table_set and child != parent:
            if child not in edges[parent]:
                edges[parent].add(child)
                incoming[child] += 1

    ready = sorted([name for name, degree in incoming.items() if degree == 0])
    ordered = []

    while ready:
        current = ready.pop(0)
        ordered.append(current)
        for neighbor in sorted(edges[current]):
            incoming[neighbor] -= 1
            if incoming[neighbor] == 0:
                ready.append(neighbor)
        ready.sort()

    # If cycles remain, process remaining tables alphabetically.
    if len(ordered) < len(table_set):
        remaining = sorted(table_set - set(ordered))
        ordered.extend(remaining)

    return ordered

def discover_table_pairs(pg_cursor, sql_cursor, pg_schema="public", sql_schema="dbo"):
    """Find tables that exist in both PostgreSQL and SQL Server, in dependency order."""
    pg_cursor.execute(
        """
        SELECT table_name
        FROM information_schema.tables
        WHERE table_schema = %s AND table_type = 'BASE TABLE'
        """,
        (pg_schema,),
    )
    pg_tables = {row[0] for row in pg_cursor.fetchall()}

    sql_cursor.execute(
        """
        SELECT t.name
        FROM sys.tables AS t
        JOIN sys.schemas AS s ON t.schema_id = s.schema_id
        WHERE s.name = ?
        """,
        (sql_schema,),
    )
    sql_tables = {row[0] for row in sql_cursor.fetchall()}

    common_tables = sorted(pg_tables & sql_tables)
    ordered_tables = dependency_order(pg_cursor, common_tables, schema_name=pg_schema)

    return [(f"{pg_schema}.{name}", f"{sql_schema}.{name}") for name in ordered_tables]

def source_columns(pg_cursor, source_table):
    """Return ordered source columns from PostgreSQL information_schema."""
    schema_name, table_name = parse_pg_table_name(source_table)
    pg_cursor.execute(
        """
        SELECT column_name
        FROM information_schema.columns
        WHERE table_schema = %s AND table_name = %s
        ORDER BY ordinal_position
        """,
        (schema_name, table_name),
    )
    return [row[0] for row in pg_cursor.fetchall()]

def migrate_table(pg_cursor, sql_cursor, source_table, dest_table):
    # The destination defines the authoritative column order for positional bulkcopy().
    dest_columns, identity = table_columns(sql_cursor, dest_table)
    if not dest_columns:
        raise RuntimeError(
            f"No destination columns found for {dest_table}. "
            "Make sure the destination table exists before migration."
        )

    src_columns = source_columns(pg_cursor, source_table)
    if not src_columns:
        raise RuntimeError(
            f"No source columns found for {source_table}. "
            "Check the source table name and schema."
        )

    # Load only columns present on both sides and keep destination column order.
    src_column_set = set(src_columns)
    load_columns = [c for c in dest_columns if c in src_column_set]
    if not load_columns:
        raise RuntimeError(
            f"No shared columns between {source_table} and {dest_table}."
        )

    source_schema, source_name = parse_pg_table_name(source_table)
    select_query = sql.SQL("SELECT {cols} FROM {schema}.{table}").format(
        cols=sql.SQL(", ").join(sql.Identifier(c) for c in load_columns),
        schema=sql.Identifier(source_schema),
        table=sql.Identifier(source_name),
    )
    pg_cursor.execute(select_query)

    copied = 0
    while True:
        batch = pg_cursor.fetchmany(10000)
        if not batch:
            break
        # Serialize JSONB or array values (dict/list) for nvarchar(max) columns.
        rows = [
            tuple(json.dumps(v) if isinstance(v, (dict, list)) else v for v in row)
            for row in batch
        ]
        # keep_identity preserves source primary keys so foreign keys still line up.
        result = sql_cursor.bulkcopy(
            dest_table,
            rows,
            batch_size=10000,
            keep_identity=identity in load_columns,
        )
        copied += result["rows_copied"]
    return copied

pg_cursor = pg_conn.cursor()
sql_cursor = sql_conn.cursor()

# Leave TABLE_MAPPINGS as None to migrate every table that exists in both schemas.
# To migrate only selected tables, replace None with explicit mappings.
TABLE_MAPPINGS = None

if TABLE_MAPPINGS is None:
    tables = discover_table_pairs(pg_cursor, sql_cursor, pg_schema="public", sql_schema="dbo")
else:
    tables = TABLE_MAPPINGS

if not tables:
    raise RuntimeError(
        "No shared tables found between source and destination schemas. "
        "Check schema names and table creation on SQL Server."
    )

print(f"Migrating {len(tables)} table(s)...")
for source_table, dest_table in tables:
    count = migrate_table(pg_cursor, sql_cursor, source_table, dest_table)
    print(f"{dest_table}: copied {count} rows")

# bulkcopy() bypasses constraint checks, so foreign keys are left untrusted.
# Re-validate each table to mark them trusted and surface any orphaned rows.
for _, dest_table in tables:
    dest_schema, dest_name = parse_sql_table_name(dest_table)
    sql_cursor.execute(
        f"ALTER TABLE [{dest_schema}].[{dest_name}] WITH CHECK CHECK CONSTRAINT ALL"
    )
sql_conn.commit()

pg_conn.close()
sql_conn.close()

Secara default, skrip ini memigrasikan setiap tabel yang ada di (publicPostgreSQL) dan dbo (SQL Server), yang diurutkan berdasarkan dependensi kunci asing. Atur TABLE_MAPPINGS ke daftar eksplisit jika Anda hanya ingin memigrasikan subset.

Ini mengasumsikan sumber dan tujuan menggunakan nama kolom yang sama, yang merupakan kasus biasa setelah Anda menulis ulang DDL. Pembantu menangani kolom identitas secara otomatis: keep_identity mempertahankan kunci primer sumber saat tujuan memiliki IDENTITY kolom, sehingga referensi kunci asing tetap utuh. Untuk mengizinkan SQL Server menetapkan kunci baru sebagai gantinya, kecualikan kolom identitas dari columns dan teruskan keep_identity=False.

Kunci asing dan kendala

bulkcopy() menggunakan protokol sisipan massal TDS, yang tidak memberlakukan kunci asing atau memeriksa batasan selama pemuatan. Tanpa permintaan eksplisit untuk memeriksa constraint tersebut, SQL Server mengabaikan constraint CHECK dan FOREIGN KEY selama impor bulk dan setelah itu menandainya sebagai tidak tepercaya, seperti yang dijelaskan dalam BULK INSERT. Perilaku ini memiliki dua konsekuensi praktis untuk migrasi:

  • Urutan pemuatan tidak berpengaruh. Anda dapat memuat tabel anak sebelum tabel induknya tanpa mengalami pelanggaran kunci asing. Pertahankan kunci utama dengan keep_identity=True, seperti yang dilakukan pembantu, sehingga nilai kunci induk dan anak masih cocok setelah pemuatan.
  • Batasan pada akhirnya menjadi tidak tepercaya. Setelah pemuatan massal, setiap kunci asing ditandai tidak tepercaya (sys.foreign_keys.is_not_trusted = 1) karena SQL Server tidak memverifikasinya. Langkah terakhir dalam skrip memvalidasi ulang setiap tabel yang dimuat dengan ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Langkah ini menandai batasan yang dipercaya sehingga pengoptimal kueri dapat menggunakannya, dan menampilkan data yang buruk. Jika baris anak merujuk ke baris induk yang tidak ada, pernyataan akan gagal dengan pelanggaran kendala integritas yang menyebutkan nama kendala tersebut, sehingga Anda dapat memperbaiki baris yatim sebelum sistem digunakan di produksi.

Keterbatasan

Tinjau perbedaan ini sebelum bermigrasi:

Topik PostgreSQL mssql-python / SQL Server
callproc() Dukungan Menaikkan NotSupportedError. Gunakan cursor.execute("EXECUTE ...") sebagai gantinya.
Parameter bertipe tabel (TVP) Tidak ada yang setara langsung Tidak didukung di driver saat ini. Gunakan tabel sementara atau JSON untuk parameter multi-baris.
Kolom bawaan ARRAY Dukungan Tidak ada tipe array. Gunakan tabel normal, array JSON, atau STRING_SPLIT().
LISTEN/NOTIFY Dukungan Tidak ada yang setara langsung. Gunakan Service Broker atau polling tingkat aplikasi.
COPY Streaming Dukungan Gunakan bulkcopy() untuk pemuatan data massal.
Mengembalikan baris yang dimodifikasi klausa RETURNING OUTPUT INSERTED / OUTPUT DELETED klausul dalam pernyataan DML.
Driver asinkron psycopg3 memiliki dukungan async bawaan mssql-python dukungan asinkron mengandalkan solusi sementara (pool thread).
Pencarian teks lengkap tsvector / tsquery CONTAINS() / FREETEXT() dengan indeks teks lengkap.
ORM (SQLAlchemy) Didukung sepenuhnya Didukung melalui dialek mssql-python bawaan di SQLAlchemy 2.1.0b2+ (pra-rilis).

Daftar periksa validasi

Gunakan daftar periksa ini untuk memverifikasi migrasi Anda:

  1. Ganti semua %s penanda parameter dengan ? atau %(name)s parameter.
  2. Pastikan semua %(name)s parameter masih berfungsi (kedua driver mendukung format ini).
  3. Tulis LIMIT/OFFSET ulang ke .OFFSET/FETCH NEXT
  4. Tulis RETURNING ulang ke OUTPUT INSERTED.
  5. Tulis ON CONFLICT ulang ke MERGE.
  6. Ganti SERIAL / BIGSERIAL dengan .IDENTITY
  7. BOOLEAN kolom diganti dengan bit.
  8. Ganti kolom array dengan tabel atau JSON yang dinormalisasi.
  9. Ganti JSONB operator denganJSON_VALUE() / JSON_QUERY() .
  10. Perbarui string koneksi untuk autentikasi Microsoft SQL.
  11. Uji aplikasi terhadap AdventureWorks atau skema target Anda.

Autentikasi dan penyebaran

Aplikasi PostgreSQL yang dikelola sendiri biasanya disebarkan dengan string koneksi yang berisi kata sandi, atau menggunakan .pgpass file dan PGPASSWORD variabel lingkungan. Azure Database for PostgreSQL mendukung autentikasi Microsoft Entra, jadi jika Anda sudah menggunakan autentikasi tanpa kata sandi, model identitas yang sama dibawa ke Azure SQL.

Untuk beban kerja produksi terhadap Azure SQL, gunakan identitas terkelola:

conn = mssql_python.connect(
    server="<server>.database.windows.net",
    database="AdventureWorks",
    authentication="ActiveDirectoryMSI",
    encrypt="yes"
)

Untuk pengembangan lokal dan CI, lihat Kontainer dan pengembangan lokal untuk pola penyiapan alur Docker, devcontainer, dan CI.

Konten terkait

  • Last updated on