Migre do PostgreSQL para Microsoft SQL com mssql-python

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 CONFLICT PostgreSQL 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:

  1. Tabela separada (normalizada). Melhor para dados indexados e consultáveis.
  2. Array JSON armazenado em nvarchar(max). Bom para metadados opacos.
  3. 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:

  1. Exporte o esquema. Use pg_dump --schema-only para 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 .
  2. Crie tabelas no Microsoft SQL. Execute o DDL reescrito no banco de dados de destino.
  3. Exportar dados. Use pg_dump --data-only --format=csv ou 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.
  4. 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 com ALTER 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:

  1. Substitua todos os marcadores de parâmetro %s por parâmetros ? ou %(name)s.
  2. Certifique-se de que todos %(name)s os parâmetros ainda funcionam (ambos os drivers suportam esse formato).
  3. Reescreva LIMIT/OFFSET para OFFSET/FETCH NEXT.
  4. Reescreva RETURNING para OUTPUT INSERTED.
  5. Reescreva ON CONFLICT para MERGE.
  6. Substitua SERIAL / BIGSERIAL por .IDENTITY
  7. BOOLEAN colunas substituídas por bit.
  8. Substitua as colunas do array por tabelas normalizadas ou JSON.
  9. Substitua os operadores JSONB por JSON_VALUE() / JSON_QUERY().
  10. Atualize a cadeia de conexão para autenticação SQL da Microsoft.
  11. 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.