Migrate from PostgreSQL to Microsoft SQL with mssql-python

Sok Python csapat először a PostgreSQL-t tanulja meg. Ha a munkaterhelésnek szüksége van olyan funkciókra, mint időbeli táblák, teljes MERGE szemantika vagy columnstore indexek, migrálj a Microsoft SQL-re. Ez az útmutató a fő döntéseket és kódváltozásokat tartalmazza, amelyek során egy Python alkalmazást a PostgreSQL-ről (psycopg2 vagy psycopg3 használatával) a Microsoft SQL-re lehet áthelyezni a mssql-python driverrel keresztül.

Megjegyzés:

Ha az Azure Database for PostgreSQL-ről migrálsz, mindkét szolgáltatás támogatja a Microsoft Entra hitelesítést és a menedzselt identitást. Az útmutatóban ismertetett kódmódosítások attól függetlenül alkalmazhatók, hogy a PostgreSQL-forrás önkezelt vagy az Azure-ban üzemeltetett.

Mit nyersz azzal, ha átlépsz a Microsoft SQL-re

A Microsoft SQL olyan képességeket tartalmaz, amelyek egyszerűsítik a biztonsági intézkedéseket, megfelelőséget és működést a termelési terhelésekhez. Ismerd meg ezeket a funkciókat, mielőtt elkezded a migrációt, hogy kihasználhasd őket az átmenet során:

Dinamikus adatmaszkolás és sorszintű biztonság. Maszkolja az oszlopokat azoknak a felhasználóknak, akiknek nincs szükségük teljes hozzáférésre, és korlátozza a sorok láthatóságát a biztonsági szabályzat szerint. Ezek a funkciók bármely illesztőprogrammal működnek.
Időbeli táblázatok (rendszer-verzióban). A Microsoft SQL automatikusan követi a sorelőzményeket. Nincsenek triggerek, nincsenek audit táblák, nincs alkalmazáskód.
Teljes MERGE szemantika. Egyetlen utasítás kezeli a INSERT, UPDATE és DELETE elemeket egy, a naplózási nyomvonalakhoz használt OUTPUT záradékkal. A PostgreSQL záradéka ON CONFLICT csak egyetlen feltétel esetén alkalmazható beillesztés vagy frissítés lehetőségeit tartalmazza.
Oszloptárolós indexek. Oszlopos tároló hozzáadása meglévő táblákhoz hibrid OLTP/analitikai munkaterhelésekhez. Nincs szükség külön analitikai adatbázisra.
Microsoft Entra ID-hitelesítés. Kapcsolódj menedzselt identitásokhoz, szolgáltatásvezetőkhöz vagy interaktív bejelentkezéshez. Az Azure Database for PostgreSQL támogatja a Microsoft Entra hitelesítést is, így ha már használod, az átállás egyszerű.

Az illesztőprogram telepítése

Mielőtt elkezdenéd, győződj meg róla, hogy van Python 3.10 vagy újabb, valamint egy célzott SQL adatbázisod.

SQL-adatbázis létrehozása

Létrehozni vagy csatlakozni SQL adatbázishoz az alábbi platformok egyikén:

SQL-adatbázis létrehozása percek alatt az Azure Portal használatával

A PostgreSQL illesztőprogramok külső natív könyvtárakat igényelnek.

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

A mssql-python meghajtó becsomagolja a natív rétegét. Windows-on nincs szükség külső driver-menedzserre vagy rendszercsomagokra.

pip install mssql-python

Linuxon és macOS-en telepíts egy kis rendszerkönyvtári készletet, amelyeket az Installációban dokumentáltak. Nincs ekvivalens annak, hogy pg_config vagy libpq-dev.

Kapcsolódási kód frissítése

A következő részek a kapcsolati láncok, hitelesítés, kontextuskezelők és a pooling kulcsfontosságú változtatásait tárgyalják.

Kapcsolódási karakterláncok

a psycopg2 DSN stringet vagy kulcsszava-argumentumokat használ.

import psycopg2

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

Az mssql-python a kulcsszóargumentumokat is támogatja, ami elkerüli azokat az URL-kódolási problémákat, amelyek gyakran előfordulnak az SQLAlchemy kapcsolati karakterláncaiban, amikor a jelszavak @, ; vagy {} karaktereket tartalmaznak.

import mssql_python

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

Vagy használj egy kapcsolati karakterlánc-et.

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

A kapcsolati karakterláncok kulcsszavainak teljes listáját lásd: Kapcsolati karakterláncok.

Authentication

A PostgreSQL hitelesítés általában felhasználónévre és jelszóra vonatkozó szabályokat használ pg_hba.conf . Az Azure Database for PostgreSQL szintén támogatja a Microsoft Entra hitelesítést. A Microsoft SQL több hitelesítési módot támogat egyetlen kapcsolódási kulcsszóval:

PostgreSQL megközelítés	MSSQL-python megfelelője
Felhasználónév és jelszó	`UID=...;PWD=...;`
SSL/TLS titkosítás	`Encrypt=yes;`(alapértelmezett módon engedélyezve Azure SQL-hez)
Entra auth (Azure PostgreSQL)	`Authentication=ActiveDirectoryDefault;` (jelszó nélkül)
Managed identity (Azure PostgreSQL)	`Authentication=ActiveDirectoryMSI;`
Service principal (Azure PostgreSQL)	`Authentication=ActiveDirectoryServicePrincipal;`

Helyi fejlesztéshez használható ActiveDirectoryDefault . Automatikusan láncolja az Azure CLI-t, a környezeti változókat és a kezelt identitást. Termeléshez használj egy speciális módot, például ActiveDirectoryMSI (menedzselt identitás) vagy ActiveDirectoryServicePrincipal hogy elkerüld a lassú hitelesítési láncsétát. Lásd a Microsoft Entra hitelesítést mind a hét hitelesítési módhoz.

Környezetkezelők

Mindkét meghajtó támogatja a kontextuskezelőket, de a viselkedés eltér:

A psycopg2 with conn: siker esetén véglegesíti, kivétel esetén pedig visszagörgeti a tranzakciót, de nem zárja le a kapcsolatot:

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

Az MSSQL-Python with conn: kilépéskor zárja a kapcsolatot. A nem elkötelezett munkát visszavonják:

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

Kapcsolatmegosztás

A psycopg2 kifejezetten megköveteli a kapcsolati pool beállítását és kezelését.

from psycopg2 import pool

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

A mssql-python meghajtó alapértelmezetten beépített pooling-et engedélyez. Nincs szükség beállításra.

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

Állítsd be a pool méretét, ha az alapértelmezettek nem illeszkednek a munkaterhelésedhez.

import mssql_python

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

A kapcsolatcsoport méretezésére és a kapcsolatcsoport kimerülésének elhárítására vonatkozó útmutatásért lásd a Kapcsolatcsoportok kezelése című részt.

SQL dialektusi különbségek

Az alábbi táblázat a gyakori PostgreSQL-mintákat a Transact-SQL (T-SQL) megfelelőinek felelteti meg:

PostgreSQL	SQL Server (T-SQL)	Jegyzetek
`SERIAL` / `BIGSERIAL`	`int IDENTITY(1,1)`	A Microsoft SQL automatikus növeléshez használja`IDENTITY`.
`TEXT`	`nvarchar(max)`	Unicode-hoz használom `nvarchar` . Előnyben `nvarchar(4000)` vagy rövidebb, ha az adatok engedik.
`BOOLEAN`	`bit`	PostgreSQL elfogadja `true`/`false`; A Microsoft SQL használja `1`/`0`.
`BYTEA`	`varbinary(max)`	Ugyanaz a koncepció, más név.
`JSONB`	`nvarchar(max)` JSON függvényekkel	A Microsoft SQL szövegként tárolja a JSON-t és érvényesíti .`ISJSON()` Lásd a JSON adatokat.
`TIMESTAMP WITH TIME ZONE`	`datetimeoffset`	Mindkettő eltolást tárol. Lásd : Dátumidőkezelés.
`INTERVAL`	Nincs közvetlen egyenértékű	Számoljon a(z) `DATEADD()` és `DATEDIFF()` használatával.
`ARRAY`	Nincs közvetlen egyenértékű	Használjon külön táblázatot, JSON-tömböt vagy `STRING_SPLIT()`.
`UUID`	`uniqueidentifier`	A `mssql-python` illesztőprogram natívan leképezi a(z) `uuid.UUID` elemet. Lásd : Modul konfiguráció.
`NOW()` / `CURRENT_TIMESTAMP`	`GETDATE()` vagy `SYSDATETIME()`	`SYSDATETIME()` nagyobb pontosságot ad.
`LIMIT 10 OFFSET 20`	`OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY`	`ORDER BY` záradék szükséges.
`\\|\\|` (karakterláncok összefűzése)	`+` vagy `CONCAT()`	`CONCAT()` kezeli a(z) `NULL` értékeket.
`COALESCE(a, b)`	`COALESCE(a, b)` vagy `ISNULL(a, b)`	`COALESCE` mindkettőben azonos.
`string_agg(col, ',')`	`STRING_AGG(col, ',')`	Elérhető SQL Server 2017+ verzióban.
`RETURNING id`	`OUTPUT INSERTED.id`	Használd `OUTPUT` a `INSERT`, `UPDATE`, vagy `DELETE` állításban.
`ON CONFLICT ... DO UPDATE`	`MERGE` nyilatkozat	`MERGE` egy állításban támogatja INSERT a + UPDATE + DELETE -t. Lásd: Lekérdezés újraírási minták.
`EXPLAIN ANALYZE`	`SET STATISTICS IO ON; SET STATISTICS TIME ON;`	Vagy használj végrehajtási terveket az SSMS-ben / Azure Data Studio-ban.
`\d tablename`	`sp_help 'tablename'`	Vagy kérdezze le `INFORMATION_SCHEMA.COLUMNS`.
`pg_dump`	`bcp`, `BACKUP DATABASE`	Használja a(z) `bulkcopy()` elemet az adatok programozott betöltéséhez Pythonból.

CREATE TABLE Példa

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

Lekérdezés újraírási minták

Az alábbi szakaszok a gyakori PostgreSQL lekérdezési mintákat és azok T-SQL megfelelőit mutatják be.

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 paraméter sorrendje megfordul. A Microsoft SQL a(z) OFFSET elemet a(z) FETCH NEXT elé helyezi.

Upsert (beillesztés vagy frissítés)

A PostgreSQL ON CONFLICT egyetlen korlátozáson kezeli az illesztés-frissítést:

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

A Microsoft SQL MERGE egyetlen utasításban kezeli a INSERT, UPDATE és DELETE elemeket. Használj egy USING klauzust paraméter aliasokkal:

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

Tömeges upsert műveletekhez helyezze át a sorokat egy ideiglenes táblába a(z) bulkcopy() használatával, majd onnan MERGE. További információkért lásd: Tömeges upsert előkészítő táblával.

A beszúrt azonosító lekérése

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 együtt használható a(z) INSERT, UPDATE és DELETE utasításokkal. Több oszlopot is vissza tud adni.

Paraméterjelölők

A psycopg2 a pozíciós paraméterekhez a %s, az elnevezett paraméterekhez pedig a %(name)s jelölést használja. A mssql-python illesztőprogram ? használ a pozicionális és %(name)s az elnevezett paraméterekhez:

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

Tranzakciók és automatikus elköteleződési különbségek

A PostgreSQL (psycopg2) automatikusan megnyit egy tranzakciót az első parancs végrehajtásakor, és explicit commit()-t igényel:

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

A mssql-python driver alapértelmezetten ugyanígy működik. Az automatikus véglegesítés ki van kapcsolva, és explicit módon meghívod a commit()-t:

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

Az automatikus commit engedélyezéséhez:

Psycopg2:

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

MSSQL-Python:

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

Az izolációs szintekről, a mentési pontokról és a holtpont utáni újrapróbálkozási mintákról lásd a Tranzakciókezelés című részt.

Típus szempontok

Az alábbi szakaszok a PostgreSQL és a Microsoft SQL közötti leggyakoribb típustérképi különbségeket tárgyalják.

JSON

A PostgreSQL natív JSONB indexelési és lekérdezési operátorokkal (->, ->>, @>). A Microsoft SQL a JSON-t nvarchar(max) formátumban tárolja, és funkciókat biztosít lekérdezésekhez:

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

A Pythonban mindkét megközelítés a json.dumps() elemet használja szerializáláshoz:

import json

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

Lásd a JSON-adatok című témakört a JSON-tárolási és -lekérdezési mintákra vonatkozó részletes útmutatásért.

UUID (Univerzálisan Egyedi Azonosító)

A PostgreSQL és az mssql-python is natívan képezi le a uuid.UUID elemet:

import uuid

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

A csatlakozási opcióhoz lásd a Modul konfigurációtnative_uuid.

Dátum és időzóna

A PostgreSQL a TIMESTAMPTZ értéket tároláskor UTC-re alakítja. A Microsoft SQL datetimeoffset megőrzi az eredeti elmozdulatot:

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

Ha következetes UTC tárhelyre van szükséged, konvertálj Python-ban a beillesztés előtt:

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

A teljes típusleképezés esetén lásd Dátumidő-kezelést .

Tömbök

A PostgreSQL támogatja a natív tömboszlopokat (INTEGER[], TEXT[]). A Microsoft SQL-nek nincs tömbtípusa. Gyakori alternatívák:

Külön tábla (normalizált). A legjobbak a lekérdezésre alkalmas, indexelt adatokhoz.
JSON-tömb van tárolva nvarchar(max)-ban. Jó átlátszatlan metaadatokhoz.
Vesszővel elválasztott karakterlánc a következővel: STRING_SPLIT() Egyszerű, de korlátozott.

# 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 alapértelmezés szerint UTF-8 formátumban tárolja az összes szöveget. A Microsoft SQL megkülönbözteti a varchart (kód oldalkódolás) és az nvarchar (UTF-16) között. A mssql-python meghajtó alapértelmezés szerint str néven küldi a Python értékeket, így a Unicode szöveg extra konfiguráció nélkül működik. Ha a sémád varchar oszlopokat használ, és el kell kerülnöd az implicit átalakítást, akkor használd setinputsizes() az oszloptípus megadására. A kódolási részletekért lásd a String és Unicode adatokat .

Tömeges betöltés és adatmozgás

A PostgreSQL tömeges műveletekhez használ COPY . Az MSSQL-python biztosítja 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)

Nagy fájlok esetén használj generátort, hogy elkerüld az egész fájl betöltését a memóriába:

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)

Lásd a Tömeges másolási műveletek című részt az oszlopleképezések, az identitáskezelés és a teljesítménnyel kapcsolatos tippek témájában.

Séma és adatmigráció

Ezt a megközelítést alkalmazzuk egy meglévő PostgreSQL adatbázis migrációjához:

Exportáld a sémát. A DDL lekéréséhez használja a(z) pg_dump --schema-only elemet. Az opciók részleteiért és szélügyeiért (tulajdonjog, jogosultságok, kiterjesztések és szűrés) lásd a PostgreSQL pg_dump hivatkozást. Írd át a DDL-t az SQL dialektus különbségi táblázat segítségével.
Táblázatok létrehozása Microsoft SQL-ben. Futtatd az átírt DDL-t a cél adatbázisoddal.
Adatok exportálása. Használd pg_dump --data-only --format=csv vagy kérdezz minden táblát psycopg2-vel. Nagy adathalmazok és kompatibilitási kapcsolók esetén nézd meg a PostgreSQL pg_dump dokumentációt, különösen az opciók szekciót.
Töltse be az adatokat a bulkcopy használatával. Olvasd el a céloszlopok sorrendjét a katalógusból, hogy ne kódolj egy oszloplistát táblánként, majd minden táblát streamelj a Microsoft SQL-be. Íme egy példaszkript:

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

Alapértelmezés szerint ez a szkript átmigrál minden olyan táblát, amely létezik mind public a (PostgreSQL), mind dbo az (SQL Server) rendszerében, idegen kulcsfüggőségek szerint sorrendben. Ha csak egy részhalmazt szeretnél migrálni, állítsd a TABLE_MAPPINGS értékét egyértelműen megadott listára.

Ez feltételezi, hogy a forrás és a célállomás ugyanazokat az oszlopneveket használja, ami a DDL újraírása után szokásos. A segéd automatikusan kezeli az identitásoszlopot: keep_identity megőrzi a forrás elsődleges kulcsokat, amikor a célpontnak van oszlopa IDENTITY , így az idegen kulcshivatkozások érintetlenek maradnak. Ahhoz, hogy az SQL Server ehelyett új kulcsokat rendeljen hozzá, hagyja ki az identitásoszlopot a(z) columns elemből, és adja át a(z) keep_identity=False értéket.

Idegen kulcsok és korlátok

bulkcopy() a TDS tömeges beszúrási protokollt használja, amely nem kényszeríti ki a külföldi kulcsokat vagy ellenőrzési korlátozásokat a betöltés alatt. Kifejezett kérés nélkül az SQL Server figyelmen kívül hagyja a CHECK és FOREIGN KEY megszorításokat a tömeges import során, majd ezt követően nem megbízhatóként jelöli meg őket, a BULK INSERT leírtak szerint. Ennek a viselkedésnek két gyakorlati következménye van a migrációra nézve:

A betöltési sorrend nem számít. Egy gyerektáblát a szülő előtt tölthetsz be anélkül, hogy idegen billentyűk hibáit érnéd. Őrizd meg az elsődleges kulcsokat a keep_identity=True segítségével, ahogyan a segéd is teszi, hogy a szülő- és a gyermekkulcs értékei a betöltés után is egyezzenek.
A korlátok végül nem megbízhatóvá válnak. Tömeges betöltés után minden külföldi kulcsot nem megbízhatónak (sys.foreign_keys.is_not_trusted = 1) jelölnek, mert az SQL Server nem ellenőrizte azt. A szkript utolsó lépése újravalidálja minden betöltött táblát .ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL Ez a lépés megjelöli a megbízható korlátokat, hogy a lekérdezésoptimalizáló felhasználhassa, és rossz adatokat is feltár. Ha egy gyermeksor egy hiányzó szülőre hivatkozik, az utasítás meghibásodik egy integritási korlátozás megsértésével, amely a korlátozást nevezi el, így javíthatod az elmaradt sorokat, mielőtt élővé válsz.

Limitations

Áttekintsd ezeket a különbségeket a migráció előtt:

Téma	PostgreSQL	mssql-python / SQL Server
`callproc()`	Támogatott	Emelések `NotSupportedError`. A `cursor.execute("EXECUTE ...")` használható helyette.
Táblázatértékű paraméterek (TVP-k)	Nincs közvetlen egyenértékű	A jelenlegi driverben nem támogatott. Használj temp táblákat vagy JSON-t többsoros paraméterekhez.
Őshonos `ARRAY` oszlopok	Támogatott	Nincs tömbtípus. Használj normalizált táblákat, JSON tömböket vagy `STRING_SPLIT()`.
`LISTEN`/`NOTIFY`	Támogatott	Nincs közvetlen egyenértékű. Használj Service Brokert vagy alkalmazásszintű felmérést.
`COPY` Streaming	Támogatott	Használja a `bulkcopy()` elemet tömeges adatbetöltéshez.
Módosított sorok visszatérése	`RETURNING` klauzula	`OUTPUT INSERTED` / `OUTPUT DELETED` záradék a DML-utasításokban.
Aszinkron meghajtó	`psycopg3` natív aszinkron támogatással rendelkezik	`mssql-python` Az aszinkron támogatás megoldásorientált (thread pool).
Teljes szöveges keresés	`tsvector` / `tsquery`	`CONTAINS()` / `FREETEXT()` teljes szöveges indexekkel.
ORM (SQLAlchemy)	Teljes mértékben támogatott	Támogatott az SQLAlchemy 2.1.0b2+ beépített mssql-python dialektusán keresztül (előzetes kiadás).

Érvényesítési ellenőrzőlista

Használja ezt az ellenőrzőlistát a migráció ellenőrzésére:

Cseréld le az összes %s paraméterjelölőt vagy ?%(name)s paraméterekre.
Győződj meg róla, hogy minden %(name)s paraméter működjön (mindkét meghajtó támogatja ezt a formátumot).
Írd át LIMIT/OFFSET erre: OFFSET/FETCH NEXT.
Írd át RETURNING-t OUTPUT INSERTED-re.
Írd át ON CONFLICT-t MERGE-re.
Cseréld le SERIAL / BIGSERIAL elemet IDENTITY-ra.
A(z) BOOLEAN oszlopokat bit váltotta fel.
Cseréld le a tömboszlopokat normalizált táblákra vagy JSON-ra.
A(z) JSONB operátorokat cseréld le erre: JSON_VALUE() / JSON_QUERY().
Frissítse a kapcsolati sztringet a Microsoft SQL-hitelesítéshez.
Teszteld az alkalmazást az AdventureWorks vagy a cél sémád alapján.

Hitelesítés és telepítés

Az önmenedzselt PostgreSQL alkalmazások általában jelszót tartalmazó kapcsolódási láncsorokkal telepítenek, vagy fájlokat és .pgpass környezeti változókat használnakPGPASSWORD. Az Azure Database for PostgreSQL támogatja a Microsoft Entra hitelesítést, tehát ha már jelszó nélküli hitelesítést használsz, ugyanaz az identitásmodell átvihető az Azure SQL-re.

Azure SQL elleni termelési munkaterhelésekhez használd managed identity-t:

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

A helyi fejlesztéshez és a CI-hoz lásd a Konténerek és helyi fejlesztés című részt a Docker-, devcontainer- és CI-folyamat-beállítási mintákhoz.

Visszajelzés

Hasznosnak találta ezt az oldalt?

Last updated on 2026-07-22

Migrate from PostgreSQL to Microsoft SQL with mssql-python

Mit nyersz azzal, ha átlépsz a Microsoft SQL-re

Az illesztőprogram telepítése

SQL-adatbázis létrehozása

Kapcsolódási kód frissítése

Kapcsolódási karakterláncok

Authentication

Környezetkezelők

Kapcsolatmegosztás

SQL dialektusi különbségek

CREATE TABLE Példa

Lekérdezés újraírási minták

Pagination

Upsert (beillesztés vagy frissítés)

A beszúrt azonosító lekérése

Paraméterjelölők

Tranzakciók és automatikus elköteleződési különbségek

Típus szempontok

JSON

UUID (Univerzálisan Egyedi Azonosító)

Dátum és időzóna

Tömbök

Unicode

Tömeges betöltés és adatmozgás

Séma és adatmigráció

Idegen kulcsok és korlátok

Limitations

Érvényesítési ellenőrzőlista

Hitelesítés és telepítés

Kapcsolódó tartalom

Visszajelzés

További források