Nota
O acesso a esta página requer autorização. Pode tentar iniciar sessão ou alterar os diretórios.
O acesso a esta página requer autorização. Pode tentar alterar os diretórios.
Muitas equipas de Python aprendem PostgreSQL pela primeira vez. Quando a sua carga de trabalho necessitar de funcionalidades como tabelas temporais, MERGE semântica integral ou índices columnstore, migre para o Microsoft SQL. Este guia cobre as principais decisões e alterações de código para mover uma aplicação Python do PostgreSQL (usando psycopg2 ou psycopg3) para Microsoft SQL usando o mssql-python driver.
Note
Se está a migrar do Base de Dados do Azure para PostgreSQL, ambos os serviços suportam autenticação Microsoft Entra e identidade gerida. As alterações de código neste guia aplicam-se independentemente de o seu código-fonte PostgreSQL ser autogerido ou alojado no Azure.
O que ganha ao mudar para Microsoft SQL
O Microsoft SQL inclui capacidades que simplificam a segurança, conformidade e operações para cargas de trabalho de produção. Compreenda estas funcionalidades antes de começar a migrar para que possa tirar partido delas durante a transição:
- Mascaramento dinâmico de dados e segurança ao nível da linha. Mascare as colunas para os utilizadores que não precisam de acesso total e restrinja a visibilidade das linhas de acordo com a política de segurança. Estas funcionalidades funcionam com qualquer condutor.
- Tabelas temporais (versão do sistema). O Microsoft SQL regista automaticamente o histórico de linhas. Sem gatilhos, sem tabelas de auditoria, sem código de aplicação.
- Semântica completa MERGE. Uma única instrução processa INSERT, UPDATE e DELETE com uma cláusula OUTPUT para registos de auditoria. A cláusula
ON CONFLICTdo PostgreSQL cobre apenas inserção ou atualização numa só restrição. - Índices de colunas. Adicione armazenamento colunar às tabelas existentes para cargas de trabalho híbridas OLTP/análise. Não é necessária uma base de dados analítica separada.
- Autenticação do ID Microsoft Entra. Conecte-se com identidades geridas, principais de serviço ou login interativo. O Base de Dados do Azure para PostgreSQL também suporta autenticação Microsoft Entra, por isso, se já o estiveres a usar, a transição é simples.
Instale o controlador
Antes de começares, certifica-te de que tens Python 3.10 ou posterior e uma base de dados SQL alvo.
Criar um banco de dados SQL
Criar ou ligar-se a uma base de dados SQL numa das seguintes plataformas:
Os drivers PostgreSQL requerem bibliotecas nativas externas.
# psycopg2 requires pg_config, libpq-dev, and platform-specific build tools
sudo apt-get install libpq-dev # Debian/Ubuntu
pip install psycopg2
O controlador mssql-python inclui a sua camada nativa. No Windows, não precisas de um gestor de drivers externo nem de pacotes de sistema.
pip install mssql-python
No Linux e macOS, instale um pequeno conjunto de bibliotecas do sistema documentadas na Instalação. Não existe equivalente a pg_config ou libpq-dev.
Atualizar código de ligação
As secções seguintes cobrem as principais alterações às cadeias de ligação, autenticação, gestores de contexto e pooling.
Cadeias de ligação
O psycopg2 usa uma cadeia DSN ou argumentos de palavras-chave.
import psycopg2
conn = psycopg2.connect(
host="<server>",
dbname="<database>",
user="<username>",
password="<password>"
)
O mssql-python também suporta argumentos por palavras-chave, o que evita os problemas de codificação de URLs que as cadeias de ligação SQLAlchemy frequentemente apresentam quando as palavras-passe contêm @, ;, ou {} caracteres.
import mssql_python
conn = mssql_python.connect(
server="<server>.database.windows.net",
database="<database>",
authentication="ActiveDirectoryDefault",
encrypt="yes"
)
Ou utilize uma cadeia de ligação.
conn = mssql_python.connect(
"Server=<server>.database.windows.net;"
"Database=<database>;"
"Authentication=ActiveDirectoryDefault;"
"Encrypt=yes;"
)
Para o conjunto completo de palavras-chave de cadeia de ligação, veja Connection strings.
Authentication
A autenticação PostgreSQL normalmente utiliza pg_hba.conf regras com nome de utilizador e palavra-passe. O Base de Dados do Azure para PostgreSQL também suporta autenticação Microsoft Entra. O Microsoft SQL suporta múltiplos modos de autenticação através de uma única palavra-chave de ligação:
| Abordagem PostgreSQL | Equivalente a mssql-python |
|---|---|
| Nome de utilizador e palavra-passe | UID=...;PWD=...; |
| Encriptação SSL/TLS |
Encrypt=yes;(ativado por defeito para SQL do Azure) |
| Entra auth (Azure PostgreSQL) |
Authentication=ActiveDirectoryDefault; (sem palavra-passe) |
| Identidade gerida (Azure PostgreSQL) | Authentication=ActiveDirectoryMSI; |
| Service principal (Azure PostgreSQL) | Authentication=ActiveDirectoryServicePrincipal; |
Uso ActiveDirectoryDefault para desenvolvimento local. Efetua automaticamente a sequência entre o CLI do Azure, as variáveis de ambiente e a identidade gerida. Para produção, utilize um modo específico, como ActiveDirectoryMSI (identidade gerida) ou ActiveDirectoryServicePrincipal, para evitar percorrer lentamente a cadeia de credenciais. Consulte Autenticação do Microsoft Entra para conhecer os sete modos de autenticação.
Gestores de contexto
Ambos os drivers suportam gestores de contexto, mas o comportamento é diferente:
O with conn: do psycopg2 confirma em caso de sucesso e faz rollback em caso de exceção, mas não fecha a conexão:
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
O with conn: do mssql-python fecha a conexão ao terminar. Trabalho não comprometido é revertido:
with mssql_python.connect(...) as conn:
with conn.cursor() as cursor:
cursor.execute("INSERT INTO ...")
conn.commit()
# Connection is closed here
Agrupamento de conexões
O PsyCopG2 requer a configuração explícita e gestão de um pool de ligações.
from psycopg2 import pool
connection_pool = pool.ThreadedConnectionPool(1, 10, dsn="...")
conn = connection_pool.getconn()
# ... use conn ...
connection_pool.putconn(conn)
O mssql-python driver tem o pooling incorporado ativado por defeito. Não é necessário qualquer configuração.
# Pooling is automatic. Each connect() call reuses pooled connections.
conn = mssql_python.connect(...)
Configura o tamanho do pool se os valores predefinidos não se adaptarem à tua carga de trabalho.
import mssql_python
mssql_python.pooling(max_size=20, idle_timeout=300)
Para obter orientações sobre o dimensionamento do conjunto e a resolução de problemas de esgotamento do conjunto, consulte Conjunto de ligações.
Diferenças nos dialetos SQL
A tabela seguinte mapeia padrões comuns do PostgreSQL para os seus equivalentes Transact-SQL (T-SQL):
| PostgreSQL | SQL Server (T-SQL) | Notes |
|---|---|---|
SERIAL / BIGSERIAL |
int IDENTITY(1,1) |
O Microsoft SQL utiliza IDENTITY para incremento automático. |
TEXT |
nvarchar(max) |
Utilize nvarchar para Unicode. Prefira nvarchar(4000) ou uma opção mais curta, quando os dados o permitirem. |
BOOLEAN |
bit |
PostgreSQL aceita true/false; O Microsoft SQL utiliza 1/0. |
BYTEA |
varbinary(max) |
Mesmo conceito, nome diferente. |
JSONB |
nvarchar(max) com funções JSON |
O Microsoft SQL armazena JSON como texto e valida com ISJSON(). Ver dados JSON. |
TIMESTAMP WITH TIME ZONE |
datetimeoffset |
Ambos armazenam deslocamento. Veja Gestão de datas. |
INTERVAL |
Sem equivalente direto | Calcule com DATEADD() e DATEDIFF(). |
ARRAY |
Sem equivalente direto | Use uma tabela separada, um array JSON, ou STRING_SPLIT(). |
UUID |
uniqueidentifier |
O controlador mssql-python mapeia uuid.UUID nativamente.
Ver configuração do módulo. |
NOW() / CURRENT_TIMESTAMP |
GETDATE() ou SYSDATETIME() |
SYSDATETIME() Dá maior precisão. |
LIMIT 10 OFFSET 20 |
OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY |
Requer uma cláusula ORDER BY. |
\|\| (concatenação de cadeias de caracteres) |
+ ou CONCAT() |
CONCAT() processa NULL valores. |
COALESCE(a, b) |
COALESCE(a, b) ou ISNULL(a, b) |
COALESCE é idêntico em ambos. |
string_agg(col, ',') |
STRING_AGG(col, ',') |
Disponível no SQL Server 2017+. |
RETURNING id |
OUTPUT INSERTED.id |
Utilize OUTPUT na instrução INSERT, UPDATE ou DELETE. |
ON CONFLICT ... DO UPDATE |
Declaração MERGE |
MERGE suporta INSERT + UPDATE + DELETE numa única instrução.
Veja Padrões de reescrita de consultas. |
EXPLAIN ANALYZE |
SET STATISTICS IO ON; SET STATISTICS TIME ON; |
Ou usar planos de execução no SSMS / Azure Data Studio. |
\d tablename |
sp_help 'tablename' |
Ou consultar INFORMATION_SCHEMA.COLUMNS. |
pg_dump |
bcp, BACKUP DATABASE |
Use bulkcopy() para carregamento de dados programáticos a partir de Python. |
CREATE TABLE exemplo
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
);
Padrões de reescrita de consultas
As secções seguintes mostram padrões comuns de consulta PostgreSQL e os seus equivalentes 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)
)
A ordem dos parâmetros é invertida. O Microsoft SQL coloca OFFSET antes FETCH NEXTde .
Upsert (inserir ou atualizar)
O ON CONFLICT do PostgreSQL permite operações de inserção ou atualização numa única restrição:
cursor.execute("""
INSERT INTO settings (key, value)
VALUES (%s, %s)
ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value
""", (key, value))
O MERGE Microsoft SQL trata INSERT, UPDATE, e DELETE numa única instrução. Use uma USING cláusula com pseudónimos de parâmetros:
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))
Para upserts em massa, organize as linhas numa tabela temporária usando bulkcopy(), e depois MERGE a partir dela. Para mais informações, consulte Bulk upsert com uma tabela de preparação.
Insira o 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 funciona com INSERT, UPDATE, e DELETE instruções. Pode devolver várias colunas.
Marcadores de parâmetros
O psycopg2 utiliza %s parâmetros posicionais e %(name)s parâmetros nomeados. O controlador mssql-python usa ? para parâmetros posicionais e %(name)s para parâmetros nomeados:
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}
)
Diferenças entre transações e autocommit
O PostgreSQL (psycopg2) abre uma transação automaticamente no primeiro comando e requer um texto explícito commit():
conn = psycopg2.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()
O mssql-python controlador funciona da mesma forma por defeito. O autocommit está desligado, e chamas commit() explicitamente:
conn = mssql_python.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()
Para ativar a confirmação automática:
PsyCopG2:
conn = psycopg2.connect(...)
conn.autocommit = True
mssql-python:
conn = mssql_python.connect(..., autocommit=True)
# or: conn.autocommit = True
Consulte Gestão de Transações para níveis de isolamento, pontos de restauro e padrões de repetição em caso de impasse.
Considerações de tipo
As secções seguintes abordam as diferenças mais comuns de mapeamento de tipos entre PostgreSQL e Microsoft SQL.
JSON
O PostgreSQL tem operadores nativos JSONB de indexação e consulta (->, ->>, @>). O Microsoft SQL armazena JSON como nvarchar(max) e fornece funções para consultas:
| 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)) |
Em Python, ambas as abordagens usam json.dumps() para serializar:
import json
cursor.execute(
"INSERT INTO #Settings ([key], data) VALUES (%(key)s, %(data)s)",
{"key": "config", "data": json.dumps({"theme": "dark", "lang": "en"})}
)
Consulte dados JSON para orientações completas sobre padrões de armazenamento e consulta JSON.
Identificador Único Universal (UUID)
Tanto o PostgreSQL como o mssql-python mapeiam uuid.UUID nativamente:
import uuid
cursor.execute(
"INSERT INTO #Events (EventID, Name) VALUES (%(event_id)s, %(name)s)",
{"event_id": uuid.uuid4(), "name": "signup"}
)
Consulte Configuração do módulo para a opção de ligação native_uuid.
Data, hora e fuso horário
O TIMESTAMPTZ do PostgreSQL é convertido para UTC aquando do armazenamento. O Microsoft SQL datetimeoffset preserva o offset original:
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})
Se precisar de armazenamento UTC consistente, converta em Python antes de inserir:
dt_utc = dt.astimezone(timezone.utc)
cursor.execute("INSERT INTO #Events (EventTime) VALUES (%(event_time)s)", {"event_time": dt_utc})
Consulte Tratamento de data e hora para obter o mapeamento completo dos tipos.
Matrizes
O PostgreSQL suporta colunas nativas de array (INTEGER[], TEXT[]). O Microsoft SQL não tem um tipo de array. Alternativas comuns:
- Tabela separada (normalizada). O melhor para dados indexados e consultáveis.
- Array JSON armazenado em nvarchar(max). Bom para metadados opacos.
-
Cadeia de caracteres separada por vírgulas com
STRING_SPLIT(). Simples, mas limitado.
# 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
O PostgreSQL armazena todo o texto como UTF-8 por defeito. O Microsoft SQL distingue entre varchar (codificação de página de códigos) e nvarchar (UTF-16). O mssql-python controlador envia valores de Python str como nvarchar por predefinição, pelo que o texto Unicode funciona sem configuração adicional. Se o seu esquema usa colunas varchar e precisa de evitar a conversão implícita, use setinputsizes() para especificar o tipo de coluna. Consulte String e dados Unicode para detalhes de codificação.
Carregamento em massa e movimentação de dados
O PostgreSQL utiliza COPY para operações em massa. MSSQL-Python fornece 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)
Para ficheiros grandes, use um gerador para evitar carregar o ficheiro inteiro na memória:
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)
Consulte Operações de cópia em massa para mapeamentos de colunas, gestão de identidade e dicas de desempenho.
Migração de esquemas e dados
Use esta abordagem para migrar uma base de dados PostgreSQL existente:
- Exporta o esquema. Utilize
pg_dump --schema-onlypara obter DDL. Para detalhes das opções e casos limites (propriedade, privilégios, extensões e filtragem), consulte a referência PostgreSQLpg_dump. Reescreva o DDL usando a tabela de diferenças de dialetos SQL . - Criar tabelas no Microsoft SQL. Executa o DDL reescrito na tua base de dados alvo.
- Exportar dados. Usa
pg_dump --data-only --format=csvou consulta cada tabela com psycopg2. Para conjuntos de dados grandes e switches de compatibilidade, consulte a documentação do PostgreSQLpg_dump, especialmente a secção de opções. - Carregar dados com bulkcopy. Lê a ordem das colunas de destino a partir do catálogo para não definires explicitamente uma lista de colunas para cada tabela e, em seguida, faz o streaming de cada tabela para o Microsoft SQL. Aqui está um script de exemplo:
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()
Por defeito, este script migra todas as tabelas que existem tanto em public (PostgreSQL) como em dbo (SQL Server), ordenadas por dependências de chaves estrangeiras. Defina TABLE_MAPPINGS para uma lista explícita se quiser migrar apenas um subconjunto.
Isto assume que a origem e o destino usam os mesmos nomes das colunas, o que é habitual depois de reescrever o DDL. O utilitário trata automaticamente a coluna de identidade: keep_identity preserva as chaves primárias da origem quando o destino tem uma coluna IDENTITY, para que as referências de chave estrangeira se mantenham intactas. Para permitir que o SQL Server atribua novas chaves em vez disso, exclua a coluna de identidade de columns e passe keep_identity=False.
Chaves estrangeiras e restrições
bulkcopy() utiliza o protocolo TDS bulk insert, que não impõe restrições de chave estrangeira nem verificação durante a carga. Sem um pedido expresso para as verificar, o SQL Server ignora as restrições CHECK e FOREIGN KEY durante uma importação em massa e marca-as posteriormente como não fidedignas, conforme descrito em BULK INSERT. Este comportamento tem duas consequências práticas para a migração:
- A ordem de carregamento não importa. Pode carregar uma tabela dependente antes da tabela principal sem provocar violações de chave estrangeira. Preserva as chaves primárias com
keep_identity=True, como faz o ajudante, para que os valores das chaves pai e filho continuem a coincidir após o carregamento. - As restrições acabam por não ser confiáveis. Após um carregamento em massa, cada chave estrangeira é marcada como não confiável (
sys.foreign_keys.is_not_trusted = 1) porque o SQL Server não a verificou. O passo final do script revalida todas as tabelas carregadas comALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Este passo marca as restrições de confiança para que o otimizador de consultas possa usá-las, e revela dados inadequados. Se uma linha filha referenciar um pai em falta, a instrução falha com uma violação de restrição de integridade que nomeia a restrição, por isso pode corrigir as linhas órfãs antes de entrar em produção.
Limitations
Reveja estas diferenças antes de migrar:
| Tópico | PostgreSQL | mssql-python / SQL Server |
|---|---|---|
callproc() |
Suportado | Aumentos NotSupportedError. Utilize cursor.execute("EXECUTE ...") em substituição. |
| Parâmetros com valor de tabela (TVPs) | Sem equivalente direto | Não é suportado no driver atual. Use tabelas temporárias ou JSON para parâmetros de várias linhas. |
Colunas nativas ARRAY |
Suportado | Sem tipo de matriz. Use tabelas normalizadas, arrays JSON ou STRING_SPLIT(). |
LISTEN/NOTIFY |
Suportado | Sem equivalente direto. Utilize o Service Broker ou a sondagem ao nível da aplicação. |
COPY Streaming |
Suportado | Use bulkcopy() para carregamento em massa de dados. |
| Devolver linhas modificadas | Cláusula RETURNING |
OUTPUT INSERTED
/
OUTPUT DELETED cláusula nas instruções DML. |
| Controlador assíncrono |
psycopg3 tem assíncrono nativo |
mssql-python O suporte ASSYNC é orientado para soluções alternativas (pool de threads). |
| Pesquisa em texto completo | tsvector / tsquery |
CONTAINS()
/
FREETEXT() com índices de texto completo. |
| ORM (SQLAlchemy) | Totalmente suportado | Suportado através do dialeto mssql-python incorporado em SQLAlchemy 2.1.0b2+ (pré-lançamento). |
Lista de verificação de validação
Use esta lista de verificação para verificar a sua migração:
- Substitua todos os
%smarcadores de parâmetros por parâmetros?ou%(name)s. - Garante que todos
%(name)sos parâmetros continuam a funcionar (ambos os drivers suportam este formato). - Reescrever
LIMIT/OFFSETpara .OFFSET/FETCH NEXT - Reescrever
RETURNINGparaOUTPUT INSERTED. - Reescrever
ON CONFLICTparaMERGE. - Substitua
SERIAL/BIGSERIALpor .IDENTITY -
BOOLEANcolunas substituídas por bit. - Substitua as colunas do array por tabelas normalizadas ou JSON.
- Substitua os operadores
JSONBporJSON_VALUE()/JSON_QUERY(). - Atualizar cadeia de ligação para a autenticação do Microsoft SQL.
- Teste a aplicação com o AdventureWorks ou com o esquema de destino.
Autenticação e implementação
As aplicações PostgreSQL autogeridas normalmente implementam com strings de ligação contendo palavras-passe ou utilizam .pgpass ficheiros e PGPASSWORD variáveis de ambiente. O Base de Dados do Azure para PostgreSQL suporta autenticação Microsoft Entra, por isso, se já estiver a usar autenticação sem palavra-passe, o mesmo modelo de identidade é transferido para o SQL do Azure.
Para cargas de trabalho de produção contra SQL do Azure, use a identidade gerida:
conn = mssql_python.connect(
server="<server>.database.windows.net",
database="AdventureWorks",
authentication="ActiveDirectoryMSI",
encrypt="yes"
)
Para desenvolvimento local e CI, veja Container e desenvolvimento local para padrões de configuração Docker, devcontainer e pipeline CI.