许多 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_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 带有用户名和密码的规则。 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 |
在 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
);
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 可与 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})
完整的类型映射,请参见日期时间处理。
数组
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。 有关选项细节和边缘情况(所有权、权限、扩展和过滤),请参见 PostgreSQLpg_dump参考文献。 用 SQL方言差异 表重写DDL。 - 在Microsoft SQL中创建表。 在目标数据库上执行重写后的 DDL。
- 导出数据。 使用
pg_dump --data-only --format=csv或使用 psycopg2 查询每个表。 对于大型数据集和兼容性切换,请查看PostgreSQLpg_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()
默认情况下,该脚本会迁移存在于(PostgreSQL)和public(SQL Server)中的每一张表dbo,按外键依赖排序。 如果你只想迁移其中一部分,请将 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() |
支持 | 加薪 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 方言提供支持。 |
验证清单
请使用这份清单来验证您的迁移情况:
- 将所有
%s参数标记替换为?或%(name)s参数。 - 确保所有
%(name)s参数仍然正常(两个驱动都支持此格式)。 - 重写
LIMIT/OFFSET为 。OFFSET/FETCH NEXT - 重写
RETURNING为OUTPUT INSERTED。 - 重写
ON CONFLICT为MERGE。 - 用
IDENTITY替换SERIAL/BIGSERIAL。 -
BOOLEAN列替换为 位。 - 用归一化表或 JSON 替换数列列。
- 将算符替换
JSONB为JSON_VALUE()/JSON_QUERY()。 - 更新用于 Microsoft SQL 身份验证的连接字符串。
- 用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 流水线的配置模式。