Migreren van PostgreSQL naar Microsoft SQL met mssql-python

Veel Python-teams leren eerst PostgreSQL. Wanneer je workload functies nodig heeft zoals temporele tabellen, volledige MERGE semantiek of columnstore-indexen, migreer dan naar Microsoft SQL. Deze gids behandelt de belangrijkste beslissingen en codewijzigingen voor het verplaatsen van een Python-applicatie van PostgreSQL (met psycopg2 of psycopg3) naar Microsoft SQL via de mssql-python driver.

Opmerking

Als je migreert van Azure Database for PostgreSQL, ondersteunen beide diensten Microsoft Entra-authenticatie en beheerde identiteit. De codewijzigingen in deze gids gelden ongeacht of je PostgreSQL-bron zelfbeheerd of Azure-gehost is.

Wat je wint door over te stappen naar Microsoft SQL

Microsoft SQL bevat mogelijkheden die beveiliging, compliance en operaties voor productieworkloads vereenvoudigen. Begrijp deze functies voordat je begint met migreren, zodat je er tijdens de overgang gebruik van kunt maken:

  • Dynamische datamaskering en beveiliging op rijniveau. Maskeer kolommen voor gebruikers die geen volledige toegang nodig hebben, en beperk de zichtbaarheid van rijen op basis van het beveiligingsbeleid. Deze functies werken met elke driver.
  • Temporele tabellen (systeemversiegevat). Microsoft SQL houdt automatisch de rijgeschiedenis bij. Geen triggers, geen audittabellen, geen applicatiecode.
  • Volledige MERGE semantiek. Een enkele instructie behandelt INSERT, UPDATE, en DELETE met een OUTPUT-clausule voor auditsporen. De clausule ON CONFLICT van PostgreSQL dekt alleen insert-or-update op één enkele constraint.
  • Columnstore-indexen. Voeg kolomopslag toe aan bestaande tabellen voor hybride OLTP/analytics workloads. Geen aparte analytics-database nodig.
  • Microsoft Entra ID authenticatie. Maak verbinding met beheerde identiteiten, diensthoofden of interactieve aanmeldingen. Azure Database for PostgreSQL ondersteunt ook Microsoft Entra-authenticatie, dus als je het al gebruikt, is de overgang eenvoudig.

Het stuurprogramma installeren

Voordat je begint, zorg ervoor dat je Python 3.10 of later hebt en een doel-SQL-database.

Een SQL-database maken

Maak een SQL-database aan of maak verbinding met een van de volgende platforms:

PostgreSQL-drivers vereisen externe native bibliotheken.

# psycopg2 requires pg_config, libpq-dev, and platform-specific build tools
sudo apt-get install libpq-dev  # Debian/Ubuntu
pip install psycopg2

De mssql-python driver bevat zijn native component. Op Windows heb je geen externe drivermanager of systeempakketten nodig.

pip install mssql-python

Installeer op Linux en macOS een kleine set systeembibliotheken die in Installatie zijn gedocumenteerd. Er is geen equivalent van pg_config of libpq-dev.

Update verbindingscode

De volgende secties behandelen de sleutelwijzigingen in verbindingsstrings, authenticatie, contextmanagers en pooling.

Aansluitstrengen

psycopg2 gebruikt een DSN-string of trefwoordargumenten.

import psycopg2

conn = psycopg2.connect(
    host="<server>",
    dbname="<database>",
    user="<username>",
    password="<password>"
)

mssql-python ondersteunt ook trefwoordargumenten, wat de URL-coderingsproblemen vermijdt die SQLAlchemy-verbindingsstrings vaak hebben wanneer wachtwoorden , @, of ; tekens bevatten{}.

import mssql_python

conn = mssql_python.connect(
    server="<server>.database.windows.net",
    database="<database>",
    authentication="ActiveDirectoryDefault",
    encrypt="yes"
)

Of gebruik een verbindingsreeks.

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes;"
)

Voor de volledige set verbindingsreeks-zoekwoorden, zie Connection strings.

Authentication

PostgreSQL-authenticatie gebruikt pg_hba.conf doorgaans regels met gebruikersnaam en wachtwoord. Azure Database for PostgreSQL ondersteunt ook Microsoft Entra-authenticatie. Microsoft SQL ondersteunt meerdere authenticatiemodi via één verbindingssleutelwoord:

PostgreSQL-benadering MSSQL-Python equivalent
Gebruikersnaam en wachtwoord UID=...;PWD=...;
SSL/TLS-encryptie Encrypt=yes;(standaard ingeschakeld voor Azure SQL)
Entra auth (Azure PostgreSQL) Authentication=ActiveDirectoryDefault; (wachtwoordloos)
Managed identity (Azure PostgreSQL) Authentication=ActiveDirectoryMSI;
Service principal (Azure PostgreSQL) Authentication=ActiveDirectoryServicePrincipal;

Gebruiken ActiveDirectoryDefault voor lokale ontwikkeling. Het doorloopt automatisch achtereenvolgens de Azure CLI, omgevingsvariabelen en beheerde identiteit. Voor productie gebruik je een specifieke modus zoals ActiveDirectoryMSI (managed identity) of ActiveDirectoryServicePrincipal om de trage credential chain walk te vermijden. Zie Microsoft Entra-authenticatie voor alle zeven authenticatiemodi.

Contextbeheerders

Beide drivers ondersteunen contextbeheerders, maar het gedrag verschilt:

psycopg2 voert bij succes een with conn: commit uit en bij een uitzondering een rollback, maar sluit de verbinding niet:

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

de with conn: van mssql-python sluit de verbinding bij het afsluiten ervan. Onverplicht werk wordt teruggedraaid:

with mssql_python.connect(...) as conn:
    with conn.cursor() as cursor:
        cursor.execute("INSERT INTO ...")
    conn.commit()
# Connection is closed here

Groepsgewijze verbindingen

PsyCOGG2 vereist expliciete opstelling en beheer van een verbindingspool.

from psycopg2 import pool

connection_pool = pool.ThreadedConnectionPool(1, 10, dsn="...")
conn = connection_pool.getconn()
# ... use conn ...
connection_pool.putconn(conn)

De mssql-python driver heeft standaard ingebouwde pooling ingeschakeld. Er is geen setup nodig.

# Pooling is automatic. Each connect() call reuses pooled connections.
conn = mssql_python.connect(...)

Configureer de poolgrootte als de standaardinstellingen niet bij je werklast passen.

import mssql_python

mssql_python.pooling(max_size=20, idle_timeout=300)

Voor advies over poolgrootte en het oplossen van pooluitputting, zie Connection pooling.

Verschillen in SQL-dialect

De volgende tabel brengt veelvoorkomende PostgreSQL-patronen in kaart met hun Transact-SQL (T-SQL) equivalenten:

PostgreSQL SQL Server (T-SQL) Aantekeningen
SERIAL / BIGSERIAL int IDENTITY(1,1) Microsoft SQL gebruikt IDENTITY voor autoincrement.
TEXT nvarchar(max) Gebruik nvarchar voor Unicode. Geef de voorkeur aan nvarchar(4000) of korter wanneer de gegevens dat toelaten.
BOOLEAN bit PostgreSQL accepteert true/false; Microsoft SQL gebruikt 1/0.
BYTEA varbinary(max) Zelfde concept, andere naam.
JSONB nvarchar(max) met JSON-functies Microsoft SQL slaat JSON op als tekst en valideert met ISJSON(). Zie JSON-gegevens.
TIMESTAMP WITH TIME ZONE datetimeoffset Beide slaan de offset op. Zie Verwerking van datum en tijd.
INTERVAL Geen direct equivalent Bereken met DATEADD() en DATEDIFF().
ARRAY Geen direct equivalent Gebruik een aparte tabel, een JSON-array, of STRING_SPLIT().
UUID uniqueidentifier Het mssql-python stuurprogramma brengt uuid.UUID systeemeigen in kaart. Zie Moduleconfiguratie.
NOW() / CURRENT_TIMESTAMP GETDATE() of SYSDATETIME() SYSDATETIME() geeft een hogere precisie.
LIMIT 10 OFFSET 20 OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY Vereist een ORDER BY clausule.
\|\| (tekenreeksconcatenatie) + of CONCAT() CONCAT() Behandelt NULL waarden.
COALESCE(a, b) COALESCE(a, b) of ISNULL(a, b) COALESCE is in beide gevallen identiek.
string_agg(col, ',') STRING_AGG(col, ',') Beschikbaar in SQL Server 2017+.
RETURNING id OUTPUT INSERTED.id Gebruik OUTPUT in de INSERT, UPDATE, of DELETE stelling.
ON CONFLICT ... DO UPDATE MERGE verklaring MERGE ondersteunt INSERT + UPDATE + DELETE in één statement. Zie patronen voor het herschrijven van zoekopdrachten.
EXPLAIN ANALYZE SET STATISTICS IO ON; SET STATISTICS TIME ON; Of gebruik uitvoeringsplannen in SSMS / Azure Data Studio.
\d tablename sp_help 'tablename' Of voer een query uit INFORMATION_SCHEMA.COLUMNS.
pg_dump bcp, BACKUP DATABASE Gebruik bulkcopy() voor het programmatisch laden van gegevens vanuit Python.

CREATE TABLE Voorbeeld

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
);

Patronen voor het herschrijven van query's

De volgende secties tonen veelvoorkomende PostgreSQL-querypatronen en hun T-SQL-equivalenten.

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)
)

De parametervolgorde is omgekeerd. Microsoft SQL plaatst OFFSET vóór FETCH NEXT.

Upsert (invoegen of bijwerken)

PostgreSQL's ON CONFLICT ondersteunt invoegen-of-bijwerken voor één enkele constraint:

cursor.execute("""
    INSERT INTO settings (key, value)
    VALUES (%s, %s)
    ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value
""", (key, value))

Microsoft SQL MERGE verwerkt INSERT, UPDATE, en DELETE in één statement. Gebruik een USING clausule met parameteraliasen:

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))

Voor bulk-upserts plaats je de rijen in een tijdelijke tabel door te gebruiken bulkcopy(), en vervolgens MERGE vanaf deze tabel. Raadpleeg voor meer informatie Bulk upsert met een stagingtabel.

Haal de ingevoegde ID op

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 werkt met INSERT, UPDATE, en DELETE statements. Het kan meerdere kolommen teruggeven.

Parametermarkeringen

Psycopg2 gebruikt %s voor positionele parameters en %(name)s voor benoemde parameters. Het mssql-python-stuurprogramma gebruikt ? voor positionele en %(name)s voor benoemde parameters:

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}
)

Verschillen tussen transacties en autocommits

PostgreSQL (psycopg2) opent automatisch een transactie bij het eerste commando en vereist een expliciete commit():

conn = psycopg2.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()

De mssql-python driver werkt standaard op dezelfde manier. Autocommit staat uit, en je roept commit() expliciet:

conn = mssql_python.connect(...)
cursor = conn.cursor()
cursor.execute("INSERT INTO ...")
conn.commit()

Om autocommit in te schakelen:

Psycopg2:

conn = psycopg2.connect(...)
conn.autocommit = True

mssql-python:

conn = mssql_python.connect(..., autocommit=True)
# or: conn.autocommit = True

Zie Transactiebeheer voor isolatieniveaus, savepoints en deadlock-herpogingspatronen.

Type-overwegingen

De volgende secties behandelen de meest voorkomende verschillen in typemapping tussen PostgreSQL en Microsoft SQL.

JSON

PostgreSQL heeft een native JSONB functie met indexerings- en query-operatoren (->, ->>, @>). Microsoft SQL slaat JSON op als nvarchar(max) en biedt functies voor het queryen:

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))

In Python worden beide benaderingen gebruikt json.dumps() om te serialiseren:

import json

cursor.execute(
    "INSERT INTO #Settings ([key], data) VALUES (%(key)s, %(data)s)",
    {"key": "config", "data": json.dumps({"theme": "dark", "lang": "en"})}
)

Zie JSON-gegevens voor volledige richtlijnen over JSON-opslag en querypatronen.

UUID (universeel unieke identificator)

Zowel PostgreSQL als mssql-python ondersteunen uuid.UUID standaard:

import uuid

cursor.execute(
    "INSERT INTO #Events (EventID, Name) VALUES (%(event_id)s, %(name)s)",
    {"event_id": uuid.uuid4(), "name": "signup"}
)

Zie Moduleconfiguratie voor de native_uuid verbindingsoptie.

Datumtijd en tijdzone

PostgreSQL TIMESTAMPTZ converteert naar UTC op opslag. Microsoft SQL's datetimeoffset behoudt de oorspronkelijke offset:

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})

Als je consistente UTC-opslag nodig hebt, converteer dan in Python voordat je het invoegt:

dt_utc = dt.astimezone(timezone.utc)
cursor.execute("INSERT INTO #Events (EventTime) VALUES (%(event_time)s)", {"event_time": dt_utc})

Zie Afhandeling van datum en tijd voor de volledige typetoewijzing.

Reeksen

PostgreSQL ondersteunt native array-kolommen (INTEGER[], TEXT[]). Microsoft SQL heeft geen array-type. Veelvoorkomende alternatieven:

  1. Aparte tabel (genormaliseerd). Het beste voor opzoekbare, geïndexeerde data.
  2. JSON-array opgeslagen in nvarchar(max). Goed voor ondoorzichtige metadata.
  3. Door komma's gescheiden string met STRING_SPLIT(). Simpel maar beperkt.
# Option 1: Normalized table
cursor.execute("INSERT INTO #ProductTags (ProductID, Tag) VALUES (%(product_id)s, %(tag)s)", {"product_id": 1, "tag": "electronics"})
cursor.execute("INSERT INTO #ProductTags (ProductID, Tag) VALUES (%(product_id)s, %(tag)s)", {"product_id": 1, "tag": "sale"})

# Option 2: JSON array
import json
tags = json.dumps(["electronics", "sale"])
cursor.execute("INSERT INTO #Products (Name, Tags) VALUES (%(name)s, %(tags)s)", {"name": "Widget", "tags": tags})

Unicode

PostgreSQL slaat standaard alle tekst op als UTF-8. Microsoft SQL maakt onderscheid tussen varchar (codepaginacodering) en nvarchar (UTF-16). De mssql-python driver stuurt standaard Python-waarden str als nvarchar, dus Unicode-tekst werkt zonder extra configuratie. Als je schema varchar-kolommen gebruikt en je de impliciete conversie wilt vermijden, gebruik setinputsizes() dan het kolomtype op te geven. Zie String- en Unicode-gegevens voor coderingsdetails.

Bulkladen en gegevensverplaatsing

PostgreSQL wordt gebruikt COPY voor bulkoperaties. mssql-python biedt 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)

Voor grote bestanden gebruik je een generator om te voorkomen dat het hele bestand in het geheugen wordt geladen:

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)

Zie Bulk copy operations voor kolommappings, identiteitsafhandeling en prestatietips.

Schema- en datamigratie

Gebruik deze aanpak om een bestaande PostgreSQL-database te migreren:

  1. Exporteer het schema. Gebruik pg_dump --schema-only om DDL op te halen. Voor opties en randgevallen (eigendom, privileges, extensies en filtering), zie de PostgreSQL-referentiepg_dump. Herschrijf de DDL met de SQL-dialectverschillentabel .
  2. Maak tabellen aan in Microsoft SQL. Voer de herschreven DDL uit tegen je doeldatabase.
  3. Gegevens exporteren. Gebruik pg_dump --data-only --format=csv of raadpleeg elke tabel met psycopg2. Voor grote datasets en compatibiliteitsswitches, bekijk de PostgreSQL-documentatiepg_dump, vooral de optiessectie.
  4. Laad gegevens met bulkcopy. Lees de volgorde van de bestemmingskolom uit de catalogus zodat je niet per tabel een kolomlijst hardcodeert, en stream elke tabel vervolgens naar Microsoft SQL. Hier volgt een voorbeeldscript:
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()

Standaard migreert dit script elke tabel die bestaat in zowel public (PostgreSQL) als dbo (SQL Server), geordend op afhankelijkheden van vreemde sleutels. Stel TABLE_MAPPINGS in op een expliciete lijst als je slechts een deel wilt migreren.

Dit gaat ervan uit dat de bron en bestemming dezelfde kolomnamen gebruiken, wat gebruikelijk is nadat je de DDL hebt herschreven. De helper verwerkt automatisch de identiteitskolom: keep_identity bewaart de bronprimaire sleutels wanneer de bestemming een IDENTITY kolom heeft, zodat vreemde sleutelreferenties intact blijven. Om SQL Server in plaats daarvan nieuwe sleutels te laten toewijzen, sluit je de identitykolom uit van columns en geef je keep_identity=False door.

Vreemde sleutels en beperkingen

bulkcopy() gebruikt het TDS-bulkinsertprotocol, dat tijdens het laden geen foreign key- of controlebeperkingen afdwingt. Zonder een expliciet verzoek om ze te controleren, negeert SQL Server CHECK- en FOREIGN KEY-beperkingen tijdens een bulkimport en markeert deze daarna als niet-vertrouwd, zoals beschreven in BULK INSERT. Dit gedrag heeft twee praktische gevolgen voor migratie:

  • De load order maakt niet uit. Je kunt een kindtabel laden vóór de oudertabel zonder foreign key-schendingen te veroorzaken. Behoud de primaire sleutels met keep_identity=True, zoals de helper doet, zodat ouder- en kindsleutelwaarden na het laden nog steeds overeenkomen.
  • Beperkingen worden uiteindelijk niet-vertrouwd. Na een bulkload wordt elke vreemde sleutel als niet vertrouwd gemarkeerd (sys.foreign_keys.is_not_trusted = 1) omdat SQL Server deze niet heeft geverifieerd. De laatste stap in het script valideert elke geladen tabel opnieuw met ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Deze stap markeert de vertrouwde constraints zodat de query-optimizer ze kan gebruiken, en het brengt slechte data naar voren. Als een kindrij een ontbrekende ouder verwijst, faalt de instructie met een integriteitsbeperkingsschending die de beperking benoemt, zodat je de weesrijen kunt herstellen voordat je live gaat.

Limitations

Bekijk deze verschillen voordat je migreert:

Onderwerp PostgreSQL mssql-python / SQL Server
callproc() Supported Verhogingen NotSupportedError. Gebruik in plaats daarvan cursor.execute("EXECUTE ...").
Parameters met tabelwaarde (TVP's) Geen direct equivalent Niet ondersteund in de huidige driver. Gebruik tijdelijke tabellen of JSON voor parameters met meerdere rijen.
Standaard-ARRAYkolommen Supported Geen arraytype. Gebruik genormaliseerde tabellen, JSON-arrays, of STRING_SPLIT().
LISTEN/NOTIFY Supported Geen direct equivalent. Gebruik Service Broker of polling op applicatieniveau.
COPY Streaming Supported Gebruik bulkcopy() voor het in bulk laden van gegevens.
Terugkeren van aangepaste rijen RETURNING-clausule OUTPUT INSERTED / OUTPUT DELETED clausule in DML-verklaringen.
Asynchroon stuurprogramma psycopg3 heeft ingebouwde asynchrone ondersteuning mssql-python Asynchrone ondersteuning is gericht op workarounds (threadpool).
Zoeken in volledige tekst tsvector / tsquery CONTAINS() / FREETEXT() met volledige tekstindexen.
ORM (SQLAlchemy) Volledig ondersteund Ondersteund via het ingebouwde mssql-python-dialect in SQLAlchemy 2.1.0b2+ (pre-release).

Validatiecontrolelijst

Gebruik deze checklist om je migratie te verifiëren:

  1. Vervang alle %s parametermarkeringen door ?- of %(name)s-parameters.
  2. Zorg ervoor dat alle %(name)s parameters nog steeds werken (beide drivers ondersteunen dit formaat).
  3. Herschrijf LIMIT/OFFSET naar .OFFSET/FETCH NEXT
  4. Herschrijf RETURNING naar OUTPUT INSERTED.
  5. Herschrijf ON CONFLICT naar MERGE.
  6. Vervang SERIAL / BIGSERIAL door IDENTITY.
  7. BOOLEAN kolommen vervangen door bit.
  8. Vervang arraykolommen door genormaliseerde tabellen of JSON.
  9. Vervang JSONB operatoren door JSON_VALUE() / JSON_QUERY().
  10. Update verbindingsreeks voor Microsoft SQL-authenticatie.
  11. Test de toepassing ten opzichte van AdventureWorks of uw doelschema.

Authenticatie en implementatie

Zelfbeheerde PostgreSQL-applicaties worden doorgaans ingezet met verbindingsstrings die wachtwoorden bevatten, of gebruiken .pgpass bestanden en PGPASSWORD omgevingsvariabelen. Azure Database for PostgreSQL ondersteunt Microsoft Entra-authenticatie, dus als je al wachtwoordloze authenticatie gebruikt, wordt hetzelfde identiteitsmodel overgenomen naar Azure SQL.

Voor productieworkloads tegen Azure SQL, gebruik managed identity:

conn = mssql_python.connect(
    server="<server>.database.windows.net",
    database="AdventureWorks",
    authentication="ActiveDirectoryMSI",
    encrypt="yes"
)

Voor lokale ontwikkeling en CI, zie Container en lokale ontwikkeling voor Docker-, devcontainer- en CI-pijplijnopzetpatronen.

Verwante onderwerpen

  • Last updated on