使用 mssql-python 從 PostgreSQL 遷移到 Microsoft SQL

許多 Python 團隊會先學習 PostgreSQL。當你的工作負載需要像是時序資料表、完整MERGE語意或欄位儲存索引等功能時，可以遷移到 Microsoft SQL。本指南涵蓋將 Python 應用程式從 PostgreSQL（使用 psycopg2 或 psycopg3）遷移到 Microsoft SQL mssql-python 時，使用驅動程式的主要決策與程式碼變更。

Note

如果你是從適用於 PostgreSQL 的 Azure 資料庫遷移過來，這兩個服務都支援 Microsoft Entra 認證和管理身份認證。本指南中的程式碼變更，無論你的 PostgreSQL 原始碼是自我管理還是 Azure 托管，都適用。

你從 Microsoft SQL 得到什麼

Microsoft SQL 包含簡化生產工作負載安全、合規與操作的功能。在開始遷移前了解這些功能，以便在轉換過程中善用：

動態資料遮蔽與列級安全。 為不需要完全存取權限的使用者遮罩欄位，並以安全政策限制列可見性。這些功能適用於任何驅動程式。
時序表（系統版本化）。 Microsoft SQL 會自動追蹤資料列歷史。沒有觸發器，沒有稽核表，沒有應用程式代碼。
完整的 MERGE 語義。 單一陳述即可處理 INSERT、UPDATE 和 DELETE，並使用 OUTPUT 子句建立稽核軌跡。 PostgreSQL 的 ON CONFLICT 子句僅涵蓋單一限制的插入或更新。
資料行存放區索引。 在現有資料表中新增欄位儲存，以支援混合 OLTP/分析工作負載。不需要另外的分析資料庫。
Microsoft Entra ID 認證。 連接管理身份、服務主體或互動式登入。適用於 PostgreSQL 的 Azure 資料庫也支援 Microsoft Entra 認證，如果你已經在使用，轉換過程相當簡單。

安裝驅動程式

開始之前，請確保你有 Python 3.10 或更新版本，並有一個目標 SQL 資料庫。

建立 SQL 資料庫

在以下平台建立或連接 SQL 資料庫：

使用 Azure 入口網站在幾分鐘內建立 SQL 資料庫

PostgreSQL 驅動程式需要外部原生函式庫。

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

驅動程式 mssql-python 會將原生層捆綁起來。在 Windows 上，你不需要外接驅動程式管理器或系統套件。

pip install mssql-python

在 Linux 和 macOS 上，請安裝少量系統函式庫，相關說明請參閱安裝一節。沒有相當於 pg_config 或 libpq-dev 的對應項目。

更新連線代碼

以下章節將涵蓋連線字串、認證、上下文管理器與池化的主要變更。

連線字串

psycopg2 使用 DSN 字串或關鍵字參數。

import psycopg2

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

mssql-python 也支援關鍵字參數，避免了 SQLAlchemy 連線字串中密碼包含 @、或;{}字元時常見的 URL 編碼問題。

import mssql_python

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

或使用連接字串。

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

關於完整的連接字串關鍵字集合，請參見 Connection strings。

Authentication

PostgreSQL 認證通常使用 pg_hba.conf 包含使用者名稱和密碼的規則。適用於 PostgreSQL 的 Azure 資料庫也支援 Microsoft Entra 認證。 Microsoft SQL 支援透過單一連線關鍵字實現多種認證模式：

PostgreSQL 做法	MSSQL-Python 等價物
使用者名稱與密碼	`UID=...;PWD=...;`
SSL/TLS 加密	`Encrypt=yes;`（預設啟用於 Azure SQL）
Entra auth （Azure PostgreSQL）	`Authentication=ActiveDirectoryDefault;` （無密碼）
Managed identity （Azure PostgreSQL）	`Authentication=ActiveDirectoryMSI;`
Service principal （Azure PostgreSQL）	`Authentication=ActiveDirectoryServicePrincipal;`

使用 ActiveDirectoryDefault 進行本機開發。它會自動串連 Azure CLI、環境變數和管理身份。在生產環境中，請使用特定模式，例如 ActiveDirectoryMSI（受控身分識別）或 ActiveDirectoryServicePrincipal，以避免緩慢的認證憑證鏈逐一嘗試。請參閱 Microsoft Entra 認證，了解所有七種認證模式。

情境管理器

兩個驅動程式都支援上下文管理器，但行為有所不同：

psycopg2 的 with conn: 會在成功時提交，並在發生例外時回滾，但不會關閉連線：

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: 在結束時會關閉連線。未承諾的工作會被回滾：

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

連線池化

Psycopg2 需要明確設定和管理連線池。

from psycopg2 import pool

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

驅動程式 mssql-python 預設已啟用內建池化功能。不需要任何設定。

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

如果預設值不符合你的工作負載，請設定池大小。

import mssql_python

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

關於池大小及池耗故障排除的指引，請參見「連線池化」。

SQL 方言差異

下表將常見的 PostgreSQL 模式對應到其 Transact-SQL（T-SQL）等效格式：

PostgreSQL	SQL Server （T-SQL）	Notes
`SERIAL` / `BIGSERIAL`	`int IDENTITY(1,1)`	Microsoft SQL 會用`IDENTITY`來自動遞增。
`TEXT`	`nvarchar(max)`	使用 `nvarchar` 表示 Unicode。若資料允許，請優先使用 `nvarchar(4000)` 或更短的長度。
`BOOLEAN`	`bit`	PostgreSQL 接受`true`/`false`;Microsoft SQL 使用 `1`/`0`.
`BYTEA`	`varbinary(max)`	概念相同，名字不同。
`JSONB`	`nvarchar(max)` 帶有 JSON 函式	Microsoft SQL 將 JSON 儲存為文字，並使用 `ISJSON()` 進行驗證。請參閱 JSON 資料。
`TIMESTAMP WITH TIME ZONE`	`datetimeoffset`	兩者都儲存偏移量。請參見日期時間處理。
`INTERVAL`	沒有直接對應	使用 `DATEADD()` 和 `DATEDIFF()` 進行計算。
`ARRAY`	沒有直接對應	使用獨立的資料表、JSON 陣列或 `STRING_SPLIT()`。
`UUID`	`uniqueidentifier`	`mssql-python`驅動程式可原生映射`uuid.UUID`。詳見模組配置。
`NOW()` / `CURRENT_TIMESTAMP`	`GETDATE()` 或 `SYSDATETIME()`	`SYSDATETIME()` 提供更高的精度。
`LIMIT 10 OFFSET 20`	`OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY`	需要條款 `ORDER BY` 。
`\\|\\|`（字串串接）	`+` 或 `CONCAT()`	`CONCAT()` 處理 `NULL` 數值。
`COALESCE(a, b)`	`COALESCE(a, b)` 或 `ISNULL(a, b)`	`COALESCE` 兩者皆相同。
`string_agg(col, ',')`	`STRING_AGG(col, ',')`	可在 SQL Server 2017+ 中取得。
`RETURNING id`	`OUTPUT INSERTED.id`	在 `INSERT`、`UPDATE` 或 `DELETE` 陳述式中使用 `OUTPUT`。
`ON CONFLICT ... DO UPDATE`	`MERGE` 陳述	`MERGE` 支援在單一陳述式中使用 INSERT + UPDATE + DELETE。請參見查詢重寫模式。
`EXPLAIN ANALYZE`	`SET STATISTICS IO ON; SET STATISTICS TIME ON;`	或者在 SSMS / Azure Data Studio 中使用執行計畫。
`\d tablename`	`sp_help 'tablename'`	或者查詢 `INFORMATION_SCHEMA.COLUMNS`。
`pg_dump`	`bcp`、`BACKUP DATABASE`	使用 `bulkcopy()` 從 Python 以程式方式載入資料。

CREATE TABLE 範例

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
);

查詢重寫模式

以下章節展示常見的 PostgreSQL 查詢模式及其 T-SQL 對應。

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)
)

參數順序相反。 Microsoft SQL 將 OFFSET 放在 FETCH NEXT 之前。

Upsert（插入或更新）

PostgreSQL ON CONFLICT 在單一限制上處理插入或更新：

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

Microsoft SQL MERGE 在一個陳述句中處理 INSERT、 UPDATE和 DELETE 。使用帶有參數別名的 USING 子句：

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))

對於大量 upsert 作業，請先使用 bulkcopy() 將資料列暫存於暫存資料表中，然後再從該資料表使用 MERGE。如需更多資訊，請參閱「使用暫存資料表進行大量 upsert」。

請輸入身分證

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 適用於 INSERT、 UPDATE和 DELETE 陳述。可回傳多個欄位。

參數標記

PsyCopg2 使用 %s 位置參數與 %(name)s 命名參數。 mssql-python 驅動程式針對位置參數使用 ?，針對具名參數使用 %(name)s：

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}
)

交易與自動提交的差異

PostgreSQL（psycopg2）會在執行第一個指令時自動開啟交易作業，並需要明確的 commit()：

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

mssql-python驅動程式預設運作方式相同。自動提交已關閉，而且你明確呼叫 commit()：

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

啟用自動提交：

psycopg2:

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

MSSQL-python：

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

請參見交易管理，了解隔離層級、存檔點及死鎖重試模式。

類型考量

以下章節將介紹 PostgreSQL 與 Microsoft SQL 之間最常見的型別映射差異。

JSON

PostgreSQL 原生具備JSONB索引與查詢運算子（->，， ->>@>）。 Microsoft SQL 將 JSON 儲存為 nvarchar（max），並提供以下查詢函式：

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))`

在 Python 中，兩種方法都用於json.dumps()序列化：

import json

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

請參閱 JSON 資料以獲得完整的 JSON 儲存與查詢模式指引。

通用唯一識別碼 (UUID)

PostgreSQL 和 mssql-python 都能原生映射 uuid.UUID ：

import uuid

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

請參見模組配置以了解 native_uuid 連接選項。

日期時間與時區

PostgreSQL TIMESTAMPTZ 在儲存時會轉換成 UTC。 Microsoft SQL datetimeoffset 保留了原始偏移量：

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})

如果你需要穩定的 UTC 儲存，請先用 Python 轉換再插入：

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

完整型態映射請參見 Datetime 處理。

陣列

PostgreSQL 支援原生陣列欄位（INTEGER[]， TEXT[]）。 Microsoft SQL 沒有陣列類型。常見的替代方案：

獨立表格 （正規化）。最適合用於可查詢且已建立索引的資料。
JSON 陣列儲存在 nvarchar（max） 中。對於不透明的元資料來說很不錯。
以逗號分隔的字串，含有 STRING_SPLIT()。簡單但有限。

# 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})

Unicode

PostgreSQL 預設所有文字以 UTF-8 格式儲存。 Microsoft SQL 區分 varchar（字碼頁編碼）與 nvarchar（UTF-16）。 mssql-python驅動程式預設以 str 形式傳送 Python 值，因此 Unicode 文字無需額外設定即可運作。如果你的結構模式使用 varchar 欄位，且需要避免隱含轉換，請用 setinputsizes() 來指定欄位類型。關於編碼細節，請參見字串與 Unicode 資料。

批量載入與資料移動

PostgreSQL 用於 COPY 批量運算。 MSSQL-Python 提供 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)

對於大型檔案，請使用產生器以避免將整個檔案載入記憶體：

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)

請參閱批量複製操作以了解欄位映射、身份處理及效能提示。

架構與資料遷移

使用此方法來遷移現有的 PostgreSQL 資料庫：

匯出綱要。 用 pg_dump --schema-only 來取得 DDL。關於選項細節與邊緣情況（擁有權、權限、擴充功能及過濾），請參閱 PostgreSQL pg_dump 參考文獻。用 SQL 方言差異表重寫 DDL。
在 Microsoft SQL 中建立資料表。 在目標資料庫上執行重寫後的 DDL。
匯出資料。 使用 pg_dump --data-only --format=csv，或使用 psycopg2 查詢各個資料表。對於大型資料集和相容性切換，請參考 PostgreSQL pg_dump 文件，特別是選項部分。
用批量複製載入資料。 從目錄讀取目標欄位順序，避免為每個資料表將欄位清單硬式編碼，然後以串流方式將各資料表寫入 Microsoft SQL 資料庫。以下是範例文稿：

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()

預設情況下，此腳本會依外鍵相依順序遷移所有存在public於（PostgreSQL）與 dbo （SQL Server）中的表格。如果你只想遷移子集，請設定 TABLE_MAPPINGS 為明確的清單。

這是假設來源和目的地使用相同的欄位名稱，這也是重寫 DDL 後常見的情況。輔助工具會自動處理身份欄位： keep_identity 當目的方有欄位 IDENTITY 時，會保留來源主鍵，確保外鍵參考保持完整。若要讓 SQL Server 指派新鍵，請將身份欄位排除於 columns 中並傳遞 keep_identity=False。

外鍵與約束

bulkcopy() 使用 TDS 批量插入協定，該協定在載入過程中不強制外鍵或檢查約束。如果未明確要求檢查這些條件約束，SQL Server 在進行大量匯入時會忽略 CHECK 和 FOREIGN KEY 條件約束，並在之後將其標記為不受信任，如 BULK INSERT 所述。這種行為對移民有兩個實際後果：

載入順序不重要。 你可以在父表之前載入子表，而不會違反外鍵限制。像輔助程式所做的那樣，使用 keep_identity=True 保留主鍵，這樣在載入後父項與子項的鍵值仍然一致。
限制最終變得不值得信任。 批次載入後，每個外金鑰都會被標記為不可信（sys.foreign_keys.is_not_trusted = 1），因為 SQL Server 沒有驗證。腳本的最後一步會重新驗證每個載入的資料表。ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL 此步驟標記受信任的限制條件，讓查詢優化器能使用，並顯示出不良資料。如果子資料列參照不存在的父資料列，該陳述式會因違反完整性約束而失敗，並在錯誤訊息中指出該約束名稱，因此你可以在正式上線前修正這些孤兒資料列。

局限性

遷移前請先檢視這些差異：

主題	PostgreSQL	mssql-python / SQL Server
`callproc()`	Supported	加薪 `NotSupportedError`。請改用 `cursor.execute("EXECUTE ...")`。
表格值參數（TVP）	沒有直接對應	目前驅動程式不支援。多列參數可以用暫存表格或 JSON 來設定。
原生 `ARRAY` 欄	Supported	沒有陣列類型。使用正規化資料表、JSON 陣列或 `STRING_SPLIT()`。
`LISTEN`/`NOTIFY`	Supported	沒有直接等值物。可以使用 Service Broker 或應用程式層級輪詢。
`COPY` 串流	Supported	使用 `bulkcopy()` 來大量載入資料。
返回修改過的行	`RETURNING` 子句	`OUTPUT INSERTED` / `OUTPUT DELETED` DML 聲明中的條款。
非同步驅動程式	`psycopg3` 具備原生非同步支援	`mssql-python` 非同步支援偏向以變通作法為主（執行緒池）。
全文搜尋	`tsvector` / `tsquery`	`CONTAINS()` / `FREETEXT()` 附有全文索引。
ORM（SQLAlchemy）	完全支援	透過 SQLAlchemy 2.1.0b2+ 中內建的 mssql-python 方言提供支援（預先發行版）。

驗證檢查清單

請使用這份清單來驗證你的遷移：

將所有 %s 參數標記換成 ? 或 %(name)s 參數。
確保所有 %(name)s 參數都正常（兩個驅動程式都支援此格式）。
重寫LIMIT/OFFSET為。OFFSET/FETCH NEXT
重寫 RETURNING 為 OUTPUT INSERTED。
重寫 ON CONFLICT 為 MERGE。
將替換 SERIAL / BIGSERIAL 為 IDENTITY。
BOOLEAN 欄已取代為位元。
將陣列欄位替換為正規化資料表或 JSON。
將運算子替換 JSONB 為 JSON_VALUE() / JSON_QUERY()。
更新 Microsoft SQL 驗證的連線字串。
用 AdventureWorks 或你的目標架構測試應用程式。

認證與部署

自我管理的 PostgreSQL 應用程式通常部署時會使用包含密碼的連線字串，或使用 .pgpass 檔案和 PGPASSWORD 環境變數。適用於 PostgreSQL 的 Azure 資料庫支援 Microsoft Entra 認證，如果你已經使用無密碼認證，相同的身份模型也會套用到 Azure SQL。

對 Azure SQL 的生產工作負載，請使用受控識別：

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

關於本地開發與 CI，請參閱容器與本地開發，了解 Docker、開發容器及 CI 管線設定模式。

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-22

使用 mssql-python 從 PostgreSQL 遷移到 Microsoft SQL

你從 Microsoft SQL 得到什麼

安裝驅動程式

建立 SQL 資料庫

更新連線代碼

連線字串

Authentication

情境管理器

連線池化

SQL 方言差異

CREATE TABLE 範例

查詢重寫模式

Pagination

Upsert（插入或更新）

請輸入身分證

參數標記

交易與自動提交的差異

類型考量

JSON

通用唯一識別碼 (UUID)

日期時間與時區

陣列

Unicode

批量載入與資料移動

架構與資料遷移

外鍵與約束

局限性

驗證檢查清單

認證與部署

相關內容

意見反應

其他資源