Observação
O acesso a essa página exige autorização. Você pode tentar entrar ou alterar diretórios.
O acesso a essa página exige autorização. Você pode tentar alterar os diretórios.
Muitas equipes de Python aprendem PostgreSQL pela primeira vez. Quando sua carga de trabalho exigir recursos como tabelas temporais, semântica MERGEcompleta ou índices columnstore, migre para o Microsoft SQL. Este guia cobre as principais decisões e alterações de código para mover um aplicativo Python do PostgreSQL (usando psycopg2 ou psycopg3) para Microsoft SQL usando o mssql-python driver.
Note
Se você está migrando do Banco de Dados do Azure para PostgreSQL, ambos os serviços suportam autenticação Microsoft Entra e identidade gerenciada. As alterações de código neste guia se aplicam independentemente de sua fonte PostgreSQL ser autogerenciada ou hospedada no Azure.
O que você ganha ao migrar para o Microsoft SQL
O Microsoft SQL inclui capacidades que simplificam a segurança, conformidade e operações para cargas de trabalho de produção. Entenda esses recursos antes de começar a migrar para poder aproveitá-los durante a transição:
- Mascaramento dinâmico de dados e segurança em nível de linha. Mascare colunas para usuários que não precisam de acesso total e restrinja a visibilidade das linhas com base na política de segurança. Esses recursos funcionam com qualquer motorista.
- Tabelas temporais (versão do sistema). O Microsoft SQL rastreia automaticamente o histórico de linhas. Sem gatilhos, sem tabelas de auditoria, sem código de aplicação.
- Semântica MERGE completa. Uma única instrução manipula INSERT, UPDATE e DELETE com uma cláusula OUTPUT para registros de auditoria. A cláusula do
ON CONFLICTPostgreSQL cobre apenas inserção ou atualização em uma única restrição. - Índices de colunas. Adicione armazenamento colunar às tabelas existentes para cargas de trabalho híbridas OLTP/análise. Não é necessário um banco de dados analítico separado.
- Autenticação do ID do Microsoft Entra. Conecte-se com identidades gerenciadas, principais de serviço ou login interativo. O Banco de Dados do Azure para PostgreSQL também suporta autenticação Microsoft Entra, então se você já está usando, a transição é simples.
Instalar o driver
Antes de começar, certifique-se de ter Python 3.10 ou posterior e um banco de dados SQL de destino.
Criar um banco de dados SQL
Crie ou conecte-se a um banco de dados SQL em uma das seguintes plataformas:
Os drivers PostgreSQL exigem 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 driver mssql-python inclui sua camada nativa. No Windows, você não precisa de um gerenciador externo de drivers ou pacotes de sistema.
pip install mssql-python
No Linux e macOS, instale um pequeno conjunto de bibliotecas do sistema documentadas no Installation. Não existe equivalente a pg_config ou libpq-dev.
Atualizar código de conexão
As seções a seguir abordam as principais mudanças em strings de conexão, autenticação, gerenciadores de contexto e pooling.
Cadeias de conexão
Psycopg2 usa uma string 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 de palavras-chave, o que evita os problemas de codificação de URLs que as strings de conexão SQLAlchemy frequentemente apresentam quando as senhas contêm @, ;, ou {} caracteres.
import mssql_python
conn = mssql_python.connect(
server="<server>.database.windows.net",
database="<database>",
authentication="ActiveDirectoryDefault",
encrypt="yes"
)
Ou use uma cadeia de conexã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 conexão, veja Connection strings.
Authentication
A autenticação PostgreSQL normalmente usa pg_hba.conf regras com nome de usuário e senha. Banco de Dados do Azure para PostgreSQL também suporta autenticação Microsoft Entra. O Microsoft SQL suporta múltiplos modos de autenticação por meio de uma única palavra-chave de conexão:
| Abordagem PostgreSQL | Equivalente MSSQL-Python |
|---|---|
| Nome de usuário e senha | UID=...;PWD=...; |
| Criptografia SSL/TLS |
Encrypt=yes;(ativado por padrão para SQL do Azure) |
| Entra auth (Azure PostgreSQL) |
Authentication=ActiveDirectoryDefault; (sem senha) |
| Managed identity (Azure PostgreSQL) | Authentication=ActiveDirectoryMSI; |
| Service principal (Azure PostgreSQL) | Authentication=ActiveDirectoryServicePrincipal; |
Use ActiveDirectoryDefault para desenvolvimento local. Ela se encadeia automaticamente através do CLI do Azure, variáveis de ambiente e identidade gerenciada. Para produção, use um modo específico como ActiveDirectoryMSI (identidade gerenciada) ou ActiveDirectoryServicePrincipal para evitar a lenta caminhada da cadeia de credenciais. Consulte autenticação do Microsoft Entra para os sete modos de autenticação.
Gerenciadores de contexto
Ambos os drivers suportam gerentes de contexto, mas o comportamento é diferente:
O psycopg2 with conn: efetua commit 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: mssql-python fecha a conexão ao sair. Trabalhos não comprometidos são revertidos:
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
PsyCopG2 exige a configuração explícita e gerenciamento de um pool de conexão.
from psycopg2 import pool
connection_pool = pool.ThreadedConnectionPool(1, 10, dsn="...")
conn = connection_pool.getconn()
# ... use conn ...
connection_pool.putconn(conn)
O driver mssql-python tem pooling interno habilitado por padrão. Não é necessário configuração.
# Pooling is automatic. Each connect() call reuses pooled connections.
conn = mssql_python.connect(...)
Configure o tamanho do pool se os padrões não se encaixarem na sua carga de trabalho.
import mssql_python
mssql_python.pooling(max_size=20, idle_timeout=300)
Para orientações sobre o dimensionamento do pool e a solução de problemas de esgotamento do pool, consulte Pool de conexões.
Diferenças no diaaleto SQL
A tabela a seguir mapeia padrões comuns do PostgreSQL para seus equivalentes Transact-SQL (T-SQL):
| PostgreSQL | SQL Server (T-SQL) | Notes |
|---|---|---|
SERIAL / BIGSERIAL |
int IDENTITY(1,1) |
O Microsoft SQL usa IDENTITY para autoincremento. |
TEXT |
nvarchar(max) |
Use nvarchar para Unicode. Prefira nvarchar(4000) ou algo mais curto quando os dados permitirem. |
BOOLEAN |
bit |
PostgreSQL aceita true/false; O Microsoft SQL usa 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(). Veja dados JSON. |
TIMESTAMP WITH TIME ZONE |
datetimeoffset |
Ambos armazenam offset. Veja Gerenciamento de datas. |
INTERVAL |
Nenhum equivalente direto | Calcule com DATEADD() e DATEDIFF(). |
ARRAY |
Nenhum equivalente direto | Use uma tabela separada, um array JSON ou STRING_SPLIT(). |
UUID |
uniqueidentifier |
O driver mssql-python mapeia uuid.UUID nativamente.
Veja 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 |
Exige uma ORDER BY cláusula. |
\|\| (concat de string) |
+ ou CONCAT() |
CONCAT() lida com valores NULL. |
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 |
Use OUTPUT na instrução INSERT, UPDATE ou DELETE. |
ON CONFLICT ... DO UPDATE |
Instrução MERGE |
MERGE suporta INSERT + UPDATE + DELETE em uma única instrução.
Veja Padrões de reescrita de consultas. |
EXPLAIN ANALYZE |
SET STATISTICS IO ON; SET STATISTICS TIME ON; |
Ou use planos de execução no SSMS / Azure Data Studio. |
\d tablename |
sp_help 'tablename' |
Ou executar a consulta 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 seções a seguir mostram padrões comuns de consulta PostgreSQL e seus equivalentes em 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. Microsoft SQL coloca OFFSET antes FETCH NEXTde .
Upsert (insira ou atualize)
O ON CONFLICT PostgreSQL lida com inserção ou atualização com uma ú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 em uma única instrução. Use uma USING cláusula com aliases 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 em uma tabela temporária usando bulkcopy(), e depois MERGE a partir dela. Para mais informações, veja 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 retornar múltiplas colunas.
Marcadores de parâmetro
Psycopg2 usa %s para parâmetros posicionais e %(name)s para parâmetros nomeados. O driver 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
PostgreSQL (psycopg2) abre uma transação automaticamente no primeiro comando e requer uma commit() explícita:
conn = psycopg2.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()
O mssql-python driver funciona da mesma forma por padrão. O autocommit está desativado, e você chama commit() explicitamente:
conn = mssql_python.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()
Para ativar o autocommit:
Psycopg2:
conn = psycopg2.connect(...)
conn.autocommit = True
MSSQL-Python:
conn = mssql_python.connect(..., autocommit=True)
# or: conn.autocommit = True
Veja Gerenciamento de transações para níveis de isolamento, pontos de restauração e padrões de nova tentativa após deadlock.
Considerações de tipo
As seções a seguir abordam as diferenças mais comuns de mapeamento de tipos entre PostgreSQL e Microsoft SQL.
JSON
O PostgreSQL possui 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"})}
)
Veja dados JSON para orientações completas sobre padrões de armazenamento e consulta JSON.
Identificador Único Universal (UUID)
Tanto o PostgreSQL quanto 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 na opção de conexão native_uuid.
Data, hora e fuso horário
No PostgreSQL, TIMESTAMPTZ é convertido para UTC ao armazenar. O datetimeoffset do Microsoft SQL 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 você 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})
Veja Tratamento de data e hora para ver o mapeamento completo de tipos.
Matrizes
PostgreSQL suporta colunas nativas de array (INTEGER[], TEXT[]). O Microsoft SQL não tem um tipo de array. Alternativas comuns:
- Tabela separada (normalizada). Melhor para dados indexados e consultáveis.
- Array JSON armazenado em nvarchar(max). Bom para metadados opacos.
-
Cadeia separada por vírgulas com
STRING_SPLIT(). Simples, mas com limitações.
# 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 padrão. O Microsoft SQL distingue entre varchar (codificação de página de código) e nvarchar (UTF-16). O mssql-python driver envia valores em Python str como nvarchar por padrão, então o texto Unicode funciona sem configuração extra. Se seu esquema usa colunas varchar e você precisa evitar a conversão implícita, use setinputsizes() para especificar o tipo de coluna. Veja String e dados Unicode para detalhes de codificação.
Carregamento em massa e movimentação de dados
O PostgreSQL usa 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 arquivos grandes, use um gerador para evitar carregar o arquivo 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)
Veja Operações de cópia em massa para mapeamentos de colunas, tratamento de identidade e dicas de desempenho.
Migração de esquemas e dados
Use essa abordagem para migrar um banco de dados PostgreSQL existente:
- Exporte o esquema. Use
pg_dump --schema-onlypara obter DDL. Para detalhes das opções e casos limites (propriedade, privilégios, extensões e filtragem), veja a referência PostgreSQLpg_dump. Reescreva o DDL usando a tabela de diferenças de dialetos SQL . - Crie tabelas no Microsoft SQL. Execute o DDL reescrito no banco de dados de destino.
- Exportar dados. Use
pg_dump --data-only --format=csvou consulte cada tabela com psycopg2. Para grandes conjuntos de dados e switches de compatibilidade, revise a documentação do PostgreSQLpg_dump, especialmente a seção de opções. - Carregue dados usando bulkcopy. Leia a ordem das colunas de destino no catálogo para não codificar uma lista de colunas por tabela e depois faça fluxo de cada tabela para o Microsoft SQL. Este é um exemplo de script:
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 padrão, esse script migra todas as tabelas existentes tanto public em (PostgreSQL) quanto dbo em (SQL Server), ordenadas por dependências de chave estrangeira. Defina TABLE_MAPPINGS para uma lista explícita se quiser migrar apenas um subconjunto.
Isso pressupõe que a origem e o destino usam os mesmos nomes das colunas, o que é o caso usual após você reescrever o DDL. O ajudante lida automaticamente com a coluna de identidade: keep_identity preserva as chaves primárias de origem quando o destino tem uma IDENTITY coluna, para que as referências de chave estrangeira permaneçam intactas. Para permitir que o SQL Server atribua novas chaves, exclua a coluna de identidade de columns e passe keep_identity=False.
Chaves e restrições estrangeiras
bulkcopy() utiliza o protocolo TDS bulk insert, que não impõe restrições de chave estrangeira ou verificação durante a carga. Sem um pedido explícito para verificá-los, o SQL Server ignora CHECK e FOREIGN KEY restrições durante uma importação em massa e as marca como não confiáveis depois, conforme descrito em BULK INSERT. Esse comportamento tem duas consequências práticas para a migração:
- A ordem de carregamento não importa. Você pode carregar uma tabela filha antes da tabela pai sem causar violações de chave estrangeira. Preserve as chaves primárias com
keep_identity=True, como faz o método auxiliar, para que os valores das chaves pai e filho continuem correspondendo após o carregamento. - As restrições acabam sendo não 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. A etapa final do script revalida toda tabela carregada comALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Essa etapa marca as restrições confiáveis para que o otimizador de consultas possa usá-las, e revela dados ruins. Se uma linha filha referenciar um pai ausente, a instrução falha com uma violação de restrição de integridade que nomeia a restrição, então você pode corrigir as linhas órfãs antes de ativar.
Limitations
Revise essas diferenças antes de migrar:
| Tópico | PostgreSQL | mssql-python / SQL Server |
|---|---|---|
callproc() |
Supported | Aumenta NotSupportedError. Use cursor.execute("EXECUTE ...") em seu lugar. |
| Parâmetros com valores de tabela (TVPs) | Nenhum equivalente direto | Não é suportado no driver atual. Use tabelas temporárias ou JSON para parâmetros de múltiplas linhas. |
Colunas nativas ARRAY |
Supported | Nenhum tipo de matriz. Use tabelas normalizadas, arrays JSON ou STRING_SPLIT(). |
LISTEN/NOTIFY |
Supported | Nenhum equivalente direto. Use o Service Broker ou pesquisas em nível de aplicação. |
COPY Streaming |
Supported | Use bulkcopy() para carregamento em massa de dados. |
| Retornando as 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 ao ASSYNC é orientado para soluções alternativas (pool de threads). |
| Pesquisa de texto completo | tsvector / tsquery |
CONTAINS()
/
FREETEXT() com índices em texto completo. |
| ORM (SQLAlchemy) | Suporte total | Suportado via o dialeto mssql-python embutido no SQLAlchemy 2.1.0b2+ (pré-lançamento). |
Lista de verificação de validação
Use esta lista de verificação para verificar sua migração:
- Substitua todos os marcadores de parâmetro
%spor parâmetros?ou%(name)s. - Certifique-se de que todos
%(name)sos parâmetros ainda funcionam (ambos os drivers suportam esse formato). - Reescreva
LIMIT/OFFSETparaOFFSET/FETCH NEXT. - Reescreva
RETURNINGparaOUTPUT INSERTED. - Reescreva
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(). - Atualize a cadeia de conexão para autenticação SQL da Microsoft.
- Teste a aplicação contra o AdventureWorks ou o seu esquema alvo.
Autenticação e implantação
Aplicações PostgreSQL autogerenciadas normalmente são implantadas com strings de conexão contendo senhas, ou utilizam .pgpass arquivos e PGPASSWORD variáveis de ambiente. O Banco de Dados do Azure para PostgreSQL oferece suporte à autenticação do Microsoft Entra, então, se você já usa autenticação sem senha, o mesmo modelo de identidade também se aplica ao SQL do Azure.
Para cargas de produção contra SQL do Azure, use identidade gerenciada:
conn = mssql_python.connect(
server="<server>.database.windows.net",
database="AdventureWorks",
authentication="ActiveDirectoryMSI",
encrypt="yes"
)
Para desenvolvimento local e CI, consulte Container e desenvolvimento local para ver padrões de configuração para Docker, devcontainer e pipeline de CI.