使用 mssql-python 从 PostgreSQL 迁移到 Microsoft SQL

许多 Python 团队首先学习 PostgreSQL。 当工作负载需要诸如时序表、完整MERGE语义或列存储索引等功能时,迁移到 Microsoft SQL。 本指南介绍了通过驱动程序mssql-python将 Python 应用从 PostgreSQL(使用 psycopg2 或 psycopg3)迁移到 Microsoft SQL 的主要决策和代码变更。

注释

如果你是从 Azure Database for PostgreSQL 迁移过来,这两个服务都支持 Microsoft Entra 认证和管理身份。 本指南中的代码更改无论你的 PostgreSQL 源是自托管还是托管 Azure,都适用。

迁移到 Microsoft SQL 的好处

Microsoft SQL 包含简化生产工作负载安全性、合规性和运营的能力。 在开始迁移之前,先了解这些功能,以便在过渡过程中充分利用它们:

  • 动态数据掩蔽行级安全 为不需要完全访问权限的用户设置屏蔽列,并通过安全策略限制行可见性。 这些功能适用于任何驱动程序。
  • 时态表(系统版本化)。 Microsoft SQL 会自动跟踪行历史。 没有触发器,没有审计表,没有应用代码。
  • 完整的 MERGE 语义。 单条语句即可处理 INSERT、UPDATE 和 DELETE,并使用 OUTPUT 子句来保留审计跟踪记录。 PostgreSQL 的 ON CONFLICT 条款仅涵盖单个约束的插入或更新。
  • 列存储索引 为现有表添加列式存储,以支持混合 OLTP/分析工作负载。 不需要单独的分析数据库。
  • Microsoft Entra ID 身份验证。 使用托管标识、服务主体或交互式登录进行连接。 Azure Database for PostgreSQL 也支持 Microsoft Entra 认证,所以如果你已经在用,过渡过程非常简单。

安装驱动程序

开始之前,确保你有Python 3.10或更高版本,并且有一个目标SQL数据库。

创建 SQL 数据库

在以下平台之一创建或连接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_configlibpq-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 带有用户名和密码的规则。 Azure Database for PostgreSQL 也支持 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) 注释
SERIAL / BIGSERIAL int IDENTITY(1,1) Microsoft SQL 使用 IDENTITY 实现自动递增。
TEXT nvarchar(max) 对 Unicode 使用 nvarchar。 如果数据允许,优先选择 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 INSERTUPDATEDELETE 语句中使用 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 bcpBACKUP 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
);

Sqlserver:

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 对应形式。

分页

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

对于批量更新插入,请先使用 bulkcopy() 将这些行暂存到临时表中,然后再通过 MERGE 从该表中进行操作。 有关详细信息,请参阅使用暂存表进行批量更新插入

获取插入的 ID

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 可与 INSERTUPDATEDELETE 语句配合使用。 它可以返回多列。

参数标记

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

完整的类型映射,请参见日期时间处理

数组

PostgreSQL 支持原生数列列(INTEGER[]TEXT[])。 Microsoft SQL 没有数组类型。 常见的替代方案:

  1. 单独的表格 (归一化)。 最适合可查询、索引化的数据。
  2. JSON 数组存储在 nvarchar(max) 中。 对于不透明的元数据来说很不错。
  3. 带有 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 数据库:

  1. 导出架构。pg_dump --schema-only 来获得DDL。 有关选项细节和边缘情况(所有权、权限、扩展和过滤),请参见 PostgreSQL pg_dump 参考文献。 用 SQL方言差异 表重写DDL。
  2. 在Microsoft SQL中创建表。 在目标数据库上执行重写后的 DDL。
  3. 导出数据。 使用 pg_dump --data-only --format=csv 或使用 psycopg2 查询每个表。 对于大型数据集和兼容性切换,请查看PostgreSQL pg_dump 文档,尤其是选项部分。
  4. 用批量复制加载数据。 从目录里读取目标列的顺序,这样你就不用在每个表里硬编码列列表,然后把每个表都流到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()

默认情况下,该脚本会迁移存在于(PostgreSQL)和public(SQL Server)中的每一张表dbo,按外键依赖排序。 如果你只想迁移其中一部分,请将 TABLE_MAPPINGS 设置为显式列表。

这假设源和目的使用相同的列名,这也是重写DDL后常见的情况。 辅助工具自动处理身份列: keep_identity 当目标有 IDENTITY 列时,会保留源主键,从而保持外键引用的完整。 要让 SQL Server 改为分配新的键值,请从 columns 中排除标识列,并传递 keep_identity=False

外键和约束

bulkcopy() 使用TDS批量插入协议,该协议在加载过程中不强制外键或检查约束。 如果没有明确请求检查这些约束,SQL Server 会在批量导入期间忽略 CHECKFOREIGN 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() 支持 加薪 NotSupportedError。 改用 cursor.execute("EXECUTE ...")
表值参数(TVP) 无直接等效项 当前驱动不支持。 多行参数可以用临时表或 JSON 来设置。
原生 ARRAY 支持 没有数组类型。 可以使用归一化表、JSON 数组或 STRING_SPLIT()
LISTEN/NOTIFY 支持 没有直接等效项。 使用 Service Broker 或应用程序级轮询。
COPY 流式传输 支持 使用 bulkcopy() 进行批量加载数据。
返回修改过的行 RETURNING 字句 OUTPUT INSERTED / OUTPUT DELETED DML声明中的条款。
异步驱动程序 psycopg3 具有原生异步 mssql-python 异步支持是基于变通(线程池)的。
全文搜索 tsvector / tsquery CONTAINS() / FREETEXT() 附有全文索引。
ORM(SQLAlchemy) 完全支持 通过 SQLAlchemy 2.1.0b2+(预发布版)中内置的 mssql-python 方言提供支持。

验证清单

请使用这份清单来验证您的迁移情况:

  1. 将所有 %s 参数标记替换为 ?%(name)s 参数。
  2. 确保所有 %(name)s 参数仍然正常(两个驱动都支持此格式)。
  3. 重写LIMIT/OFFSET为 。OFFSET/FETCH NEXT
  4. 重写 RETURNINGOUTPUT INSERTED
  5. 重写 ON CONFLICTMERGE
  6. IDENTITY替换SERIAL / BIGSERIAL
  7. BOOLEAN 列替换为
  8. 用归一化表或 JSON 替换数列列。
  9. 将算符替换 JSONBJSON_VALUE() / JSON_QUERY()
  10. 更新用于 Microsoft SQL 身份验证的连接字符串。
  11. 用AdventureWorks或你的目标模式测试应用。

认证与部署

自管理的 PostgreSQL 应用程序通常部署时包含密码的连接字符串,或使用 .pgpass 文件和 PGPASSWORD 环境变量。 Azure Database for PostgreSQL 支持 Microsoft Entra 认证,所以如果你已经在使用无密码认证,相同的身份模型也会迁移到 Azure SQL。

对于针对Azure SQL的生产工作负载,使用管理身份:

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

有关本地开发和 CI,请参阅 容器与本地开发,了解 Docker、devcontainer 和 CI 流水线的配置模式。