Migrar de PostgreSQL para Microsoft SQL com mssql-python

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

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

  1. Exporta o esquema. Utilize pg_dump --schema-only para 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 .
  2. Criar tabelas no Microsoft SQL. Executa o DDL reescrito na tua base de dados alvo.
  3. Exportar dados. Usa pg_dump --data-only --format=csv ou 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.
  4. 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 com ALTER 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:

  1. Substitua todos os %smarcadores de parâmetros por parâmetros ? ou %(name)s.
  2. Garante que todos %(name)s os parâmetros continuam a funcionar (ambos os drivers suportam este formato).
  3. Reescrever LIMIT/OFFSET para .OFFSET/FETCH NEXT
  4. Reescrever RETURNING para OUTPUT INSERTED.
  5. Reescrever ON CONFLICT para MERGE.
  6. Substitua SERIAL / BIGSERIAL por .IDENTITY
  7. BOOLEANcolunas 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. Atualizar cadeia de ligação para a autenticação do Microsoft SQL.
  11. 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.