Migration von PostgreSQL zu Microsoft SQL mit mssql-python

Viele Python-Teams lernen zuerst PostgreSQL. Wenn Ihre Arbeitslast Funktionen wie temporale Tabellen, vollständige MERGE Semantik oder Columnstore-Indizes benötigt, migrieren Sie auf Microsoft SQL. Dieser Leitfaden behandelt die wichtigsten Entscheidungen und Codeänderungen für den Umzug einer Python-Anwendung von PostgreSQL (unter Verwendung von psycopg2 oder psycopg3) auf Microsoft SQL mithilfe des Treibersmssql-python.

Was gewinnst du, wenn du zu Microsoft SQL wechselst

Microsoft SQL bietet Funktionen, die Sicherheit, Compliance und Abläufe für Produktionsarbeitslasten vereinfachen. Verstehen Sie diese Funktionen, bevor Sie mit der Migration beginnen, damit Sie sie während des Übergangs nutzen können:

• Dynamische Datenmaskierung und Sicherheit auf Zeilenebene. Maskiere Spalten für Nutzer, die keinen vollen Zugriff benötigen, und schränke die Zeilensichtbarkeit durch Sicherheitsrichtlinien ein. Diese Funktionen funktionieren mit jedem Fahrer.
• Zeitliche Tabellen (systemversioniert). Microsoft SQL verfolgt die Zeilenhistorie automatisch. Keine Trigger, keine Audit-Tabellen, kein Anwendungscode.
• Vollständige MERGE Semantik. Eine einzelne Aussage behandelt INSERT, UPDATE, und DELETE mit einer OUTPUT-Klausel für Audit-Trails. Die ON CONFLICT Klausel von PostgreSQL deckt nur das Einfügen oder Aktualisieren einer einzelnen Einschränkung ab.
• Columnstore-Indizes. Fügen Sie bestehenden Tabellen für hybride OLTP/Analytics-Workloads einen Spaltenspeicher hinzu. Keine separate Analysedatenbank erforderlich.
• Microsoft Entra ID-Authentifizierung. Verbinden Sie sich mit verwalteten Identitäten, Service-Principals oder interaktiver Anmeldung. Azure Database for PostgreSQL unterstützt ebenfalls Microsoft Entra-Authentifizierung, sodass der Übergang unkompliziert ist, wenn Sie sie bereits nutzen.

Installieren des Treibers

Bevor du anfängst, stelle sicher, dass du Python 3.10 oder später hast und eine Ziel-SQL-Datenbank.

Erstellen einer SQL-Datenbank

Erstellen oder verbinden Sie sich mit einer SQL-Datenbank auf einer der folgenden Plattformen:

PostgreSQL-Treiber benötigen 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

Der Treiber mssql-python bündelt seine native Schicht. Unter Windows brauchst du keinen externen Treibermanager oder Systempakete.

pip install mssql-python

Installieren Sie unter Linux und macOS eine kleine Anzahl von Systembibliotheken, die in Installation dokumentiert sind. Es gibt kein Äquivalent zu pg_config oder libpq-dev.

Verbindungscode aktualisieren

Die folgenden Abschnitte behandeln die Schlüsseländerungen an Verbindungszeichenfolgen, Authentifizierung, Kontextmanager und Pooling.

Verbindungszeichenfolgen

psycopg2 verwendet eine DSN-Zeichenkette oder Schlüsselwortargumente.

import psycopg2

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

mssql-python unterstützt außerdem Schlüsselwortargumente, was die URL-Codierungsprobleme vermeidet, die SQLAlchemy-Verbindungszeichenketten oft haben, wenn Passwörter , @, oder ; Zeichen enthalten{}.

import mssql_python

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

Oder verwenden Sie eine Verbindungszeichenfolge.

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

Für den vollständigen Satz der Verbindungszeichenfolge-Schlüsselwörter siehe Connection strings.

Authentifizierung

Die PostgreSQL-Authentifizierung verwendet pg_hba.conf typischerweise Regeln mit Benutzername und Passwort. Azure Database for PostgreSQL unterstützt außerdem die Microsoft Entra-Authentifizierung. Microsoft SQL unterstützt mehrere Authentifizierungsmodi über ein einziges Verbindungsschlüsselwort:

Verwenden Sie ActiveDirectoryDefault für die lokale Entwicklung. Es durchläuft automatisch Azure CLI, Umgebungsvariablen und die verwaltete Identität. Für die Produktion verwenden Sie einen bestimmten Modus wie ActiveDirectoryMSI (verwaltete Identität) oder ActiveDirectoryServicePrincipal um den langsamen Credential-Chain Walk zu vermeiden. Siehe Microsoft Entra-Authentifizierung für alle sieben Authentifizierungsmodi.

Kontextmanager

Beide Treiber unterstützen Kontextmanager, aber das Verhalten unterscheidet sich:

psycopg2s with conn: führt bei Erfolg einen Commit durch und bei einer Ausnahme ein Rollback, schließt aber die Verbindung nicht:

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

Das with conn: von mssql-python schließt die Verbindung beim Beenden. Ungebundene Arbeit wird zurückgedreht:

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

Verbindungspooling

psycopg2 erfordert eine explizite Einrichtung und Verwaltung eines Verbindungspools.

from psycopg2 import pool

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

Der Treiber mssql-python hat standardmäßig eingebautes Pooling aktiviert. Es ist keine Einrichtung nötig.

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

Konfiguriere die Poolgröße, falls die Standardeinstellungen nicht zu deiner Arbeitslast passen.

import mssql_python

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

Für Hinweise zur Poolgröße und zur Problembehebung der Pool-Erschöpfung siehe Connection Pooling.

Unterschiede im SQL-Dialekt

Die folgende Tabelle ordnet gängige PostgreSQL-Muster ihren Transact-SQL (T-SQL)-Äquivalenten zu:

CREATE TABLE Beispiel

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

Abfrage-Umschreibmuster

Die folgenden Abschnitte zeigen gängige PostgreSQL-Abfragemuster und deren T-SQL-Entsprechungen.

Paginierung

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

Die Reihenfolge der Parameter ist umgekehrt. Microsoft SQL stellt OFFSET vor FETCH NEXT.

Upsert (Einfügen oder Aktualisieren)

PostgreSQLs ON CONFLICT verarbeitet Einfüge- oder Aktualisierungsvorgänge bei nur einer 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 verwaltet INSERT, UPDATE, und DELETE in einer Anweisung. Verwenden Sie eine USING Klausel mit 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))

Für Bulk-Upserts werden die Zeilen mithilfe von bulkcopy() in einer temporären Tabelle zwischengespeichert und dann daraus mit MERGE verarbeitet. Weitere Informationen finden Sie unter Massen-Upsert mit einer Staging-Tabelle.

Setze einen Ausweis ein

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 funktioniert mit INSERT, UPDATE, und DELETE Anweisungen. Es kann mehrere Spalten zurückgeben.

Parametermarkierungen

psycopg2 verwendet %s für positionsbezogene Parameter und %(name)s für benannte Parameter. Der mssql-python Treiber verwendet ? für positionale und %(name)s für benannte Parameter:

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

Unterschiede zwischen Transaktionen und Autocommit

PostgreSQL (psycopg2) öffnet eine Transaktion automatisch beim ersten Befehl und erfordert ein explizites commit():

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

Der Treiber mssql-python funktioniert standardmäßig genauso. Autocommit ist ausgeschaltet, und du rufst commit() explizit:

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

Um Autocommit zu aktivieren:

psycopg2:

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

mssql-python:

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

Siehe Transaktionsmanagement für Isolationsstufen, Speicherpunkte und Deadlock-Wiederholungsmuster.

Typenüberlegungen

Die folgenden Abschnitte behandeln die häufigsten Unterschiede in der Typabbildung zwischen PostgreSQL und Microsoft SQL.

JSON

PostgreSQL bietet native Unterstützung für JSONB mit Indexierungs- und Abfrageoperatoren (->, ->>, @>). Microsoft SQL speichert JSON als nvarchar(max) und bietet Funktionen zum Abfragen an:

In Python verwenden beide Ansätze json.dumps() zum Serialisieren:

import json

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

Siehe JSON-Daten für vollständige Hinweise zu JSON-Speicher- und Abfragemustern.

UUID

Sowohl PostgreSQL als auch mssql-python unterstützen uuid.UUID nativ:

import uuid

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

Siehe Modulkonfiguration für die native_uuid Verbindungsoption.

Datum, Zeit und Zeitzone

PostgreSQL konvertiert TIMESTAMPTZ auf UTC im Speicher. Microsoft SQL datetimeoffset bewahrt den ursprünglichen 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})

Wenn du konsistenten UTC-Speicher brauchst, konvertiere in Python, bevor du einfügst:

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

Siehe Datetime-Handling für die vollständige Typzuordnung.

Felder

PostgreSQL unterstützt native Array-Spalten (INTEGER[], TEXT[]). Microsoft SQL hat keinen Array-Typ. Häufige Alternativen:

1. Separate Tabelle (normalisiert). Am besten für abfragbare, indexierte Daten.
2. JSON-Array in nvarchar(max) gespeichert. Gut für undurchsichtige Metadaten.
3. Durch Kommas getrennte Zeichenkette mit STRING_SPLIT(). Einfach, aber begrenzt.

# 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 speichert standardmäßig alle Texte als UTF-8. Microsoft SQL unterscheidet zwischen varchar (Codepage-Codierung) und nvarchar (UTF-16). Der Treiber mssql-python sendet standardmäßig Python-Werte str als nvarchar, sodass Unicode-Text ohne zusätzliche Konfiguration funktioniert. Wenn dein Schema Varchar-Spalten verwendet und du die implizite Umwandlung vermeiden musst, gib setinputsizes() den Spaltentyp an. Siehe String- und Unicode-Daten für Kodierungsdetails.

Massenladen und Datenbewegung

PostgreSQL verwendet COPY für Massenoperationen. mssql-python bietet 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)

Für große Dateien verwenden Sie einen Generator, um zu vermeiden, dass die gesamte Datei in den Speicher geladen wird:

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)

Siehe Massenkopieroperationen für Spaltenabbildungen, Identitätshandhabung und Performance-Tipps.

Schema- und Datenmigration

Verwenden Sie diesen Ansatz, um eine bestehende PostgreSQL-Datenbank zu migrieren:

1. Exportiere das Schema. Verwenden Sie pg_dump --schema-only, um DDL zu erhalten. Für Details zu Optionen und Randfällen (Eigentum, Privilegien, Erweiterungen und Filterung) siehe die PostgreSQL-Referenzpg_dump. Schreibe die DDL mit der SQL-Dialektunterschiedstabelle um.
2. Erstellen Sie Tabellen in Microsoft SQL. Führen Sie die umgeschriebene DDL gegen Ihre Zieldatenbank aus.
3. Exportieren von Daten. Verwende pg_dump --data-only --format=csv oder frage jede Tabelle mit psycopg2 ab. Für große Datensätze und Kompatibilitätsswitches lesen Sie die PostgreSQL-Dokumentationpg_dump, insbesondere den Optionsbereich.
4. Daten mit Bulkcopy laden. Lies die Reihenfolge der Zielspalten aus dem Katalog, damit du keine Spaltenliste pro Tabelle festkodieren musst, und streame dann jede Tabelle in Microsoft SQL. Hier ist ein Beispielskript:

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

Standardmäßig migriert dieses Skript jede Tabelle, die sowohl in public (PostgreSQL) als auch in dbo (SQL Server) vorhanden ist, in der Reihenfolge ihrer Fremdschlüsselabhängigkeiten. Setze TABLE_MAPPINGS auf eine explizite Liste, wenn du nur eine Teilmenge migrieren möchtest.

Dies setzt voraus, dass Quelle und Ziel dieselben Spaltennamen verwenden, was nach dem Umschreiben der DDL üblich ist. Der Helfer verarbeitet die Identitätsspalte automatisch: keep_identity behält die Primärschlüssel der Quelle bei, wenn das Ziel über eine IDENTITY-Spalte verfügt, damit Fremdschlüsselverweise erhalten bleiben. Um SQL Server stattdessen neue Schlüssel zuweisen zu lassen, schließen Sie die Identity-Spalte aus columns aus und geben Sie keep_identity=False an.

Fremdschlüssel und Einschränkungen

bulkcopy() verwendet das TDS-Bulk-Insert-Protokoll, das beim Laden weder Fremdschlüssel- noch Prüfbeschränkungen erzwingt. Ohne eine ausdrückliche Anforderung, sie zu überprüfen, ignoriert SQL Server beim Massenimport CHECK- und FOREIGN KEY-Constraints und markiert sie anschließend als nicht vertrauenswürdig, wie in BULK INSERT beschrieben. Dieses Verhalten hat zwei praktische Folgen für die Migration:

• Die Ladereihenfolge spielt keine Rolle. Du kannst eine Kindertabelle vor ihrem Elternteil laden, ohne auf Fremdschlüsselverstöße zu stoßen. Bewahren Sie die Primärschlüssel mit keep_identity=True, wie es der Helfer tut, damit Eltern- und Kindschlüsselwerte auch nach dem Laden übereinstimmen.
• Einschränkungen werden letztlich als nicht vertrauenswürdig eingestuft. Nach einer Massenladung wird jeder Fremdschlüssel als nicht vertrauenswürdig (sys.foreign_keys.is_not_trusted = 1) markiert, weil SQL Server ihn nicht überprüft hat. Der letzte Schritt im Skript validiert jede geladene Tabelle erneut mit ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Dieser Schritt markiert die vertrauenswürdigen Constraints, damit der Abfrageoptimierer sie verwenden kann, und zeigt fehlerhafte Daten auf. Wenn eine Kindzeile auf einen fehlenden Elternteil verweist, scheitert die Anweisung mit einer Integritätsbedingungsverletzung, die die Einschränkung benannt, sodass du die verwaisten Zeilen vor dem Live-Gehen beheben kannst.

Einschränkungen

Überprüfen Sie diese Unterschiede, bevor Sie migrieren:

Überprüfungsprüfliste

Verwenden Sie diese Checkliste, um Ihre Migration zu überprüfen:

1. Ersetzen Sie alle %s Parametermarker durch ? oder %(name)s Parameter.
2. Stellen Sie sicher, dass alle %(name)s Parameter weiterhin funktionieren (beide Treiber unterstützen dieses Format).
3. Umschreiben LIMIT/OFFSET auf .OFFSET/FETCH NEXT
4. Umschreiben RETURNING auf OUTPUT INSERTED.
5. Umschreiben ON CONFLICT auf MERGE.
6. Ersetzen Sie SERIAL / BIGSERIAL durch .IDENTITY
7. BOOLEAN Spalten wurden durch bit ersetzt.
8. Ersetzen Sie Array-Spalten durch normalisierte Tabellen oder JSON.
9. Ersetze JSONB Operatoren durchJSON_VALUE() / JSON_QUERY() .
10. Aktualisieren Sie die Verbindungszeichenfolge für die Microsoft SQL-Authentifizierung.
11. Teste die Anwendung mit AdventureWorks oder deinem Zielschema.

Authentifizierung und Bereitstellung

Selbstverwaltete PostgreSQL-Anwendungen werden typischerweise mit Verbindungszeichen, die Passwörter enthalten, bereitgestellt oder verwenden .pgpass Dateien und PGPASSWORD Umweltvariablen. Azure Database for PostgreSQL unterstützt Microsoft Entra-Authentifizierung, sodass dasselbe Identitätsmodell auf Azure SQL übertragen wird, wenn Sie bereits passwortlose Auth verwenden.

Für Produktionsworkloads gegen Azure SQL verwenden Sie Managed Identity:

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

Für lokale Entwicklung und CI siehe Container und lokale Entwicklung für Docker-, devcontainer- und CI-Pipeline-Setup-Patterns.

Verwandte Inhalte

• Installation
• Verbindungszeichenfolgen
• Microsoft Entra-Authentifizierung
• Massenkopieroperationen
• Datentypzuordnungen
• Transaktionsverwaltung
• Container und lokale Entwicklung
• Migration von SQLite