Migrera från PostgreSQL till Microsoft SQL med mssql-python

Många Python-team lär sig först PostgreSQL. När din arbetslast behöver funktioner som temporala tabeller, fullständig MERGE-semantik eller kolumnlagringsindex, migrera till Microsoft SQL. Denna guide täcker de viktigaste besluten och kodändringarna för att flytta en Python-applikation från PostgreSQL (med psycopg2 eller psycopg3) till Microsoft SQL genom att använda drivrutinenmssql-python.

Note

Om du migrerar från Azure Database for PostgreSQL stöder båda tjänsterna Microsoft Entra-autentisering och hanterad identitet. Kodändringarna i denna guide gäller oavsett om din PostgreSQL-källkod är självhanterad eller Azure-hostad.

Vad du vinner på att byta till Microsoft SQL

Microsoft SQL inkluderar funktioner som förenklar säkerhet, efterlevnad och drift för produktionsarbetsbelastningar. Förstå dessa funktioner innan du börjar migrera så att du kan dra nytta av dem under övergången:

Dynamisk datamaskering och säkerhet på radnivå. Maskera kolumner för användare som inte behöver full åtkomst, och begränsa radens synlighet enligt säkerhetspolicy. Dessa funktioner fungerar med alla drivrutiner.
Temporala tabeller (systemversionerade). Microsoft SQL spårar automatiskt radhistorik. Inga triggers, inga revisionstabeller, ingen applikationskod.
Fullständig MERGE semantik. Ett enda statement hanterar INSERT, UPDATE, och DELETE med en OUTPUT-klausul för revisionsspår. PostgreSQL:s ON CONFLICT klausul täcker endast insättning eller uppdatering på en enda begränsning.
Kolumnlagringsindex. Lägg till kolumnlagring i befintliga tabeller för hybrida OLTP/analysarbetsbelastningar. Ingen separat analysdatabas behövs.
Microsoft Entra-ID-autentisering. Koppla upp dig mot hanterade identiteter, tjänsteprinciper eller interaktiv inloggning. Azure Database for PostgreSQL stöder också Microsoft Entra-autentisering, så om du redan använder det är övergången enkel.

Installera drivrutinen

Innan du börjar, se till att du har Python 3.10 eller senare och en riktad SQL-databas.

Skapa en SQL-databas

Skapa eller koppla till en SQL-databas på en av följande plattformar:

Skapa en SQL-databas på några minuter med hjälp av Azure-portalen

PostgreSQL-drivrutiner kräver externa inbyggda bibliotek.

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

mssql-python-drivrutinen innehåller sitt interna lager. På Windows behöver du inte en extern drivrutinshanterare eller systempaket.

pip install mssql-python

På Linux och macOS, installera en liten uppsättning systembibliotek dokumenterade i Installation. Det finns inget motsvarande till pg_config eller libpq-dev.

Uppdatera anslutningskod

Följande avsnitt täcker nyckeländringar i anslutningssträngar, autentisering, kontexthanterare och poolning.

Anslutningssträngar

psycopg2 använder en DSN-sträng eller nyckelordsargument.

import psycopg2

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

mssql-python stöder också nyckelordsargument, vilket undviker de URL-kodningsproblem som SQLAlchemy-anslutningssträngar ofta har när lösenord innehåller @, ;, eller {} tecken.

import mssql_python

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

Eller använd en anslutningssträng.

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

Den fullständiga uppsättningen nyckelord för anslutningssträngar finns i Anslutningssträngar.

Authentication

PostgreSQL-autentisering använder pg_hba.conf vanligtvis regler med användarnamn och lösenord. Azure Database for PostgreSQL stöder också Microsoft Entra-autentisering. Microsoft SQL stöder flera autentiseringslägen via ett enda anslutningsnyckelord:

PostgreSQL-metoden	MSSQL-Python ekvivalent
Användarnamn och lösenord	`UID=...;PWD=...;`
SSL/TLS-kryptering	`Encrypt=yes;`(aktiverat som standard för Azure SQL)
Entra auth (Azure PostgreSQL)	`Authentication=ActiveDirectoryDefault;` (lösenordslös)
Managed identity (Azure PostgreSQL)	`Authentication=ActiveDirectoryMSI;`
Service principal (Azure PostgreSQL)	`Authentication=ActiveDirectoryServicePrincipal;`

Använd ActiveDirectoryDefault för lokal utveckling. Den går automatiskt igenom Azure CLI, miljövariabler och hanterad identitet. För produktion, använd ett specifikt läge som ActiveDirectoryMSI (managed identity) eller ActiveDirectoryServicePrincipal för att undvika den långsamma kedjevandringen av inloggningsuppgifter. Se Microsoft Entra-autentisering för alla sju autentiseringslägen.

Kontexthanterare

Båda drivrutinerna stödjer kontexthanterare, men beteendet skiljer sig åt:

Psycopg2:s satsar with conn: på framgång och rullar tillbaka vid undantag, men stänger inte kopplingen:

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

mssql-python's with conn: stänger anslutningen när den avslutas. Oengagerat arbete rullas tillbaka:

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

Anslutningspoolning

Psycopg2 kräver explicit installation och hantering av en anslutningspool.

from psycopg2 import pool

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

Drivrutinen mssql-python har inbyggd pooling aktiverad som standard. Ingen installation behövs.

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

Konfigurera poolstorleken om standardinställningarna inte passar din arbetsbelastning.

import mssql_python

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

För vägledning om dimensionering av anslutningspooler och felsökning när anslutningspoolen töms, se Anslutningspooler.

Skillnader i SQL-dialekter

Följande tabell mappar vanliga PostgreSQL-mönster till deras Transact-SQL (T-SQL) motsvarigheter:

PostgreSQL	SQL Server (T-SQL)	Noteringar
`SERIAL` / `BIGSERIAL`	`int IDENTITY(1,1)`	Microsoft SQL använder `IDENTITY` för autoincrement.
`TEXT`	`nvarchar(max)`	Använd `nvarchar` för Unicode. Föredra `nvarchar(4000)` eller kortare när data tillåter.
`BOOLEAN`	`bit`	PostgreSQL accepterar `true`/`false`; Microsoft SQL använder `1`/`0`.
`BYTEA`	`varbinary(max)`	Samma koncept, annat namn.
`JSONB`	`nvarchar(max)` med JSON-funktioner	Microsoft SQL lagrar JSON som text och validerar med `ISJSON()`. Se JSON-data.
`TIMESTAMP WITH TIME ZONE`	`datetimeoffset`	Båda lagrar offsetvärdet. Se Datetime-hantering.
`INTERVAL`	Ingen direkt motsvarighet	Beräkna med `DATEADD()` och `DATEDIFF()`.
`ARRAY`	Ingen direkt motsvarighet	Använd en separat tabell, en JSON-array, eller `STRING_SPLIT()`.
`UUID`	`uniqueidentifier`	Föraren `mssql-python` kartar `uuid.UUID` naturligt. Se Modulkonfiguration.
`NOW()` / `CURRENT_TIMESTAMP`	`GETDATE()` eller `SYSDATETIME()`	`SYSDATETIME()` ger högre precision.
`LIMIT 10 OFFSET 20`	`OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY`	Kräver en `ORDER BY` klausul.
`\\|\\|` (strängsammanfogning)	`+` eller `CONCAT()`	`CONCAT()` hanterar `NULL` värden.
`COALESCE(a, b)`	`COALESCE(a, b)` eller `ISNULL(a, b)`	`COALESCE` är identisk i båda.
`string_agg(col, ',')`	`STRING_AGG(col, ',')`	Tillgänglig i SQL Server 2017+.
`RETURNING id`	`OUTPUT INSERTED.id`	Använd `OUTPUT` i `INSERT`, `UPDATE`, eller `DELETE` satsen.
`ON CONFLICT ... DO UPDATE`	`MERGE` uttalande	`MERGE` stöder INSERT + UPDATE + DELETE i en sats. Se Frågeomskrivningsmönster.
`EXPLAIN ANALYZE`	`SET STATISTICS IO ON; SET STATISTICS TIME ON;`	Eller använd exekveringsplaner i SSMS / Azure Data Studio.
`\d tablename`	`sp_help 'tablename'`	Eller fråga `INFORMATION_SCHEMA.COLUMNS`.
`pg_dump`	`bcp`, `BACKUP DATABASE`	Används `bulkcopy()` för programmatisk dataladdning från Python.

CREATE TABLE Exempel

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

Omskrivningsmönster för frågor

Följande avsnitt visar vanliga PostgreSQL-frågemönster och deras T-SQL-motsvarigheter.

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

Parameterordningen är omvänd. Microsoft SQL sätter OFFSET före FETCH NEXT.

Upsert (infoga eller uppdatera)

PostgreSQLs ON CONFLICT hanterar infogning eller uppdatering för ett enda villkor:

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

Microsoft SQL:s MERGE hanterar INSERT, UPDATE, och DELETE i en sats. Använd en USING klausul med parameteralias:

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 upserts i bulk, mellanlagra raderna i en tillfällig tabell med hjälp av bulkcopy(), och använd sedan MERGE från den. Mer information finns i Massupsert med en mellantabell.

Få insatta 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 fungerar med INSERT, UPDATE, och DELETE satser. Den kan returnera flera kolumner.

Parametermarkörer

PsyCopg2 används %s för positionsparametrar och %(name)s för namngivna parametrar. Drivrutinen mssql-python använder ? för positionsparametrar och %(name)s för namngivna parametrar:

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

Skillnader mellan transaktion och autocommit

PostgreSQL (psycopg2) öppnar automatiskt en transaktion vid det första kommandot och kräver en explicit commit():

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

Drivrutinen mssql-python fungerar på samma sätt som standard. Autocommit är avstängt, och du ropar commit() uttryckligen:

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

För att aktivera autocommit:

Psycopg2:

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

mssql-python:

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

Se Transaktionshantering för isoleringsnivåer, sparpunkter och återförsöksmönster för deadlock.

Typöverväganden

Följande avsnitt täcker de vanligaste skillnaderna i typmappning mellan PostgreSQL och Microsoft SQL.

JSON

PostgreSQL har inbyggt JSONB indexerings- och frågeoperatorer (->, ->>, @>). Microsoft SQL lagrar JSON som nvarchar(max) och tillhandahåller funktioner för att förfråga:

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

I Python använder båda metoderna json.dumps() för att serialisera:

import json

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

Se JSON-data för fullständig vägledning om JSON-lagring och frågemönster.

universellt unik identifierare (UUID)

Både PostgreSQL och mssql-python mappas uuid.UUID nativt:

import uuid

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

Se Modulkonfiguration för anslutningsalternativetnative_uuid.

Datum, tid och tidszon

PostgreSQL konverterar TIMESTAMPTZ till UTC på lagring. Microsoft SQL:s datetimeoffset bevarar den ursprungliga offseten:

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

Om du behöver konsekvent UTC-lagring, konvertera i Python innan du sätter i:

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

Se Datetime-hantering för fullständig typmappning.

Array

PostgreSQL stöder inbyggda arraykolumner (INTEGER[], TEXT[]). Microsoft SQL har ingen arraytyp. Vanliga alternativ:

Separat tabell (normaliserad). Bäst för sökbara, indexerade data.
JSON-arrayen lagrad i nvarchar(max). Bra för ogenomskinlig metadata.
Kommaseparerad sträng med STRING_SPLIT(). Enkelt men begränsat.

# 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 lagrar all text som UTF-8 som standard. Microsoft SQL skiljer mellan varchar (kodkodning) och nvarchar (UTF-16). Drivrutinen mssql-python skickar Python-värden str som nvarchar som standard, så Unicode-text fungerar utan extra konfiguration. Om ditt schema använder varchar-kolumner och du behöver undvika den implicita konverteringen, använd setinputsizes() det för att specificera kolumntypen. Se String- och Unicode-data för kodningsdetaljer.

Bulkladdning och dataförflyttning

PostgreSQL används COPY för bulkoperationer. MSSQL-Python tillhandahåller 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 stora filer, använd en generator för att undvika att ladda hela filen i minnet:

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)

Se Bulkkopieringsoperationer för kolumnmappningar, identitetshantering och prestandatips.

Schema- och datamigration

Använd detta tillvägagångssätt för att migrera en befintlig PostgreSQL-databas:

Exportera schemat. Använd pg_dump --schema-only för att få DDL. För detaljer om alternativ och undantagsfall (ägarskap, privilegier, tillägg och filtrering), se referensen PostgreSQLpg_dump. Skriv om DDL med SQL-tabellen för dialektskillnader .
Skapa tabeller i Microsoft SQL. Kör den omskrivna DDL:n mot din måldatabas.
Exportera data. Använd pg_dump --data-only --format=csv eller fråga varje tabell med psycopg2. För stora datamängder och kompatibilitetsswitchar, granska PostgreSQL-dokumentationenpg_dump, särskilt optionssektionen.
Ladda data med bulkkopia. Läs destinationskolumnordningen från katalogen så att du inte hårdkodar en kolumnlista per tabell, och strömma sedan varje tabell in i Microsoft SQL. Här är ett exempelskript:

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

Som standard migrerar detta skript varje tabell som finns i både public (PostgreSQL) och dbo (SQL Server), ordnade efter främmande nyckelberoenden. Ställ TABLE_MAPPINGS in på en explicit lista om du vill migrera endast en delmängd.

Detta förutsätter att källan och destinationen använder samma kolumnnamn, vilket är det vanliga fallet efter att du skrivit om DDL. Hjälparen hanterar identitetskolumnen automatiskt: keep_identity bevarar källnycklarna när destinationen har en IDENTITY kolumn, så att främmande nyckelreferenser förblir intakta. För att låta SQL Server tilldela nya nycklar istället, exkludera identitetskolumnen från columns och skicka keep_identity=False.

Främmande nycklar och begränsningar

bulkcopy() använder TDS bulk insert-protokollet, som inte upprätthåller främmande nyckel- eller kontrollbegränsningar under belastningen. Utan en uttrycklig begäran om att kontrollera dem ignorerar SQL Server CHECK- och FOREIGN KEY-begränsningar vid en bulkimport och markerar dem som ej betrodda efteråt, som beskrivs i BULK INSERT. Detta beteende har två praktiska konsekvenser för migration:

Laddningsordningen spelar ingen roll. Du kan ladda en barntabell före dess förälder utan att trycka på främmande nyckelöverträdelser. Bevara primärnycklarna med keep_identity=True, som hjälparen gör, så att föräldra- och barnnyckelvärden fortfarande matchar efter laddningen.
Begränsningar blir till slut opålitliga. Efter en bulk-laddning markeras varje främmande nyckel som inte betrodd (sys.foreign_keys.is_not_trusted = 1) eftersom SQL Server inte verifierade den. Det sista steget i skriptet validerar om varje laddad tabell med ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Detta steg markerar de begränsningar som är betrodda så att frågeoptimeraren kan använda dem, och det visar felaktig data. Om en barnrad refererar till en saknad förälder, misslyckas satsen med ett integritetsbegränsningsbrott som namnger begränsningen, så du kan fixa de föräldralösa raderna innan du går live.

Limitations

Gå igenom dessa skillnader innan du migrerar:

Ämne	PostgreSQL	mssql-python / SQL Server
`callproc()`	Stöds	Höjer `NotSupportedError`. Använd `cursor.execute("EXECUTE ...")` i stället.
Tabellvärda parametrar (TVP)	Ingen direkt motsvarighet	Stöds inte i den nuvarande drivrutinen. Använd temptabeller eller JSON för flerradsparametrar.
Inhemska `ARRAY` kolonner	Stöds	Ingen arraytyp. Använd normaliserade tabeller, JSON-arrayer eller `STRING_SPLIT()`.
`LISTEN`/`NOTIFY`	Stöds	Ingen direkt motsvarighet. Använd Service Broker eller applikationsnivåpolling.
`COPY` Streaming	Stöds	Använd `bulkcopy()` för bulkdataladdning.
Returnera modifierade rader	`RETURNING`-klausul	`OUTPUT INSERTED` / `OUTPUT DELETED` klausul i DML-uttalanden.
Asynkron drivrutin	`psycopg3` har inbyggd asynk	`mssql-python` Asynkront stöd bygger på temporära lösningar (trådpool).
Fulltextsökning	`tsvector` / `tsquery`	`CONTAINS()` / `FREETEXT()` med fulltextindex.
ORM (SQLAlchemy)	Stöds fullt ut	Stöds via den inbyggda mssql-python-dialekten i SQLAlchemy 2.1.0b2+ (förhandsversion).

Checklista för verifiering

Använd denna checklista för att verifiera din migration:

Byt ut alla %s parametermarkörer med ? eller %(name)s parametrar.
Se till att alla %(name)s parametrar fortfarande fungerar (båda drivrutinerna stödjer detta format).
Skriv LIMIT/OFFSET om till .OFFSET/FETCH NEXT
Skriv RETURNING om till OUTPUT INSERTED.
Skriv ON CONFLICT om till MERGE.
Byt ut SERIAL / BIGSERIAL med .IDENTITY
BOOLEAN Kolumner ersatta med Bit.
Byt ut arraykolumner mot normaliserade tabeller eller JSON.
Byt ut JSONB operatorer mot JSON_VALUE() / JSON_QUERY().
Uppdatera reťazec pripojenia för Microsoft SQL-autentisering.
Testa applikationen mot AdventureWorks eller ditt målschema.

Autentisering och utrullning

Självhanterade PostgreSQL-applikationer distribueras vanligtvis med anslutningssträngar som innehåller lösenord, eller använder .pgpass filer och PGPASSWORD miljövariabler. Azure Database for PostgreSQL stöder Microsoft Entra-autentisering, så om du redan använder lösenordslös autentisering överförs samma identitetsmodell till Azure SQL.

För produktionsarbetsbelastningar mot Azure SQL, använd managed identity:

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

För lokal utveckling och CI, se Container och lokal utveckling för Docker-, devcontainer- och CI-pipelineuppsättningsmönster.

Feedback

Var den här sidan till hjälp?

Last updated on 2026-07-22

Migrera från PostgreSQL till Microsoft SQL med mssql-python

Vad du vinner på att byta till Microsoft SQL

Installera drivrutinen

Skapa en SQL-databas

Uppdatera anslutningskod

Anslutningssträngar

Authentication

Kontexthanterare

Anslutningspoolning

Skillnader i SQL-dialekter

CREATE TABLE Exempel

Omskrivningsmönster för frågor

Pagination

Upsert (infoga eller uppdatera)

Få insatta ID

Parametermarkörer

Skillnader mellan transaktion och autocommit

Typöverväganden

JSON

universellt unik identifierare (UUID)

Datum, tid och tidszon

Array

Unicode

Bulkladdning och dataförflyttning

Schema- och datamigration

Främmande nycklar och begränsningar

Limitations

Checklista för verifiering

Autentisering och utrullning

Relaterat innehåll

Feedback

Ytterligare resurser