Migrate from PostgreSQL to Microsoft SQL with mssql-python

Mnoho Python týmů se nejprve učí PostgreSQL. Když vaše pracovní zátěž potřebuje funkce jako časové tabulky, kompletní MERGE sémantiku nebo indexy columnstore, přejděte na Microsoft SQL. Tento průvodce pokrývá hlavní rozhodnutí a změny v kódu pro přesun Python aplikace z PostgreSQL (pomocí psycopg2 nebo psycopg3) do Microsoft SQL pomocí ovladačemssql-python.

Note

Pokud migrujete z Azure Database for PostgreSQL, obě služby podporují autentizaci Microsoft Entra a spravovanou identitu. Změny v tomto návodu platí bez ohledu na to, zda je váš zdrojový kód PostgreSQL samospravovaný nebo hostovaný v Azure.

Co získáte přechodem na Microsoft SQL

Microsoft SQL obsahuje funkce, které zjednodušují bezpečnost, dodržování předpisů a provoz pro produkční pracovní zátěže. Zjistěte si tyto funkce před začátkem migrace, abyste je mohli během přechodu využít:

Dynamické maskování dat a bezpečnost na úrovni řádků. Maskujte sloupce pro uživatele, kteří nepotřebují plný přístup, a omezte viditelnost řádků bezpečnostní politikou. Tyto funkce fungují s jakýmkoli řidičem.
Časové tabulky (systémově verzované). Microsoft SQL automaticky sleduje historii řádků. Žádné spouštěče, žádné auditní tabulky, žádný aplikační kód.
Kompletní MERGE sémantika. Jeden příkaz obsluhuje INSERT, UPDATE a DELETE s klauzulí OUTPUT pro auditní záznamy. Klauzule ON CONFLICT v PostgreSQL se vztahuje pouze na operaci vložení nebo aktualizace pro jediné omezení.
Sloupcové indexy. Přidejte sloupcové úložiště do stávajících tabulek pro hybridní OLTP/analytické zátěže. Není potřeba žádná samostatná analytická databáze.
Ověřování ID Microsoft Entra. Připojte se ke spravovaným identitám, principům služeb nebo interaktivnímu přihlášení. Azure Database for PostgreSQL také podporuje autentizaci Microsoft Entra, takže pokud ji už používáte, přechod je jednoduchý.

Instalace ovladače

Než začnete, ujistěte se, že máte Python 3.10 nebo novější a cílovou SQL databázi.

Vytvoření databáze SQL

Vytvořte nebo se připojte k SQL databázi na jedné z následujících platforem:

Vytvoření databáze SQL v řádu minut pomocí webu Azure Portal

Ovladače PostgreSQL vyžadují externí nativní knihovny.

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

Ovladač mssql-python obsahuje svou nativní vrstvu. Na Windows nepotřebujete externí správce ovladačů ani systémové balíčky.

pip install mssql-python

Na Linuxu a macOS nainstalujte malou sadu systémových knihoven zdokumentovaných v Installation. Neexistuje ekvivalent k pg_config ani k libpq-dev.

Aktualizujte kód připojení

Následující sekce pokrývají klíčové změny v řetězcích spojů, autentizaci, správcích kontextu a poolingu.

Připojovací řetězce

psycopg2 používá DSN řetězec nebo klíčové argumenty.

import psycopg2

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

mssql-python také podporuje argumenty klíčových slov, což zabraňuje problémům s kódováním URL, které se často vyskytují u řetězců spojení SQLAlchemy, když hesla obsahují @, ;, nebo {} znaky.

import mssql_python

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

Nebo použijte připojovací řetězec.

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

Pro kompletní sadu klíčových slov připojovací řetězec viz Connection strings.

Autentizace

Autentizace PostgreSQL obvykle používá pg_hba.conf pravidla s uživatelským jménem a heslem. Azure Database for PostgreSQL také podporuje autentizaci Microsoft Entra. Microsoft SQL podporuje více autentizačních režimů prostřednictvím jednoho klíčového slova spojení:

Přístup PostgreSQL	MSSQL-Python ekvivalent
Uživatelské jméno a heslo	`UID=...;PWD=...;`
SSL/TLS šifrování	`Encrypt=yes;`(ve výchozím nastavení povoleno pro Azure SQL)
Entra auth (Azure PostgreSQL)	`Authentication=ActiveDirectoryDefault;` (bez hesla)
Managed identity (Azure PostgreSQL)	`Authentication=ActiveDirectoryMSI;`
Service principal (Azure PostgreSQL)	`Authentication=ActiveDirectoryServicePrincipal;`

Používá se ActiveDirectoryDefault pro místní vývoj. Automaticky řetězí Azure CLI, proměnné prostředí a spravovanou identitu. Pro produkci použijte konkrétní režim, například ActiveDirectoryMSI (spravovaná identita) nebo ActiveDirectoryServicePrincipal, abyste se vyhnuli pomalému procházení řetězce přihlašovacích údajů. Viz Microsoft Entra autentizace pro všech sedm autentizačních režimů.

Kontextoví manažeři

Oba ovladače podporují správce kontextu, ale chování se liší:

psycopg2 při úspěchu with conn: potvrdí transakci a v případě výjimky provede rollback, ale spojení neuzavírá:

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: uzavře spojení při ukončení. Nezávazná práce se stáhne zpět:

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

Sdílení připojení

psycopg2 vyžaduje explicitní nastavení a správu poolu připojení.

from psycopg2 import pool

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

Ovladač mssql-python má ve výchozím nastavení povolené vestavěné sdružování připojení. Není potřeba žádné nastavení.

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

Pokud výchozí nastavení neodpovídá vaší pracovní zátěži, nastavte velikost poolu.

import mssql_python

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

Pokyny k dimenzování fondu připojení a řešení vyčerpání fondu připojení naleznete v části Sdružování připojení.

Rozdíly v SQL dialektech

Následující tabulka mapuje běžné vzory PostgreSQL na jejich ekvivalenty Transact-SQL (T-SQL):

PostgreSQL	SQL Server (T-SQL)	Poznámky
`SERIAL` / `BIGSERIAL`	`int IDENTITY(1,1)`	Microsoft SQL používá pro automatické zvyšování hodnot `IDENTITY`.
`TEXT`	`nvarchar(max)`	Použití `nvarchar` pro Unicode. Preferujte nebo zkracujte `nvarchar(4000)` , pokud to data dovolují.
`BOOLEAN`	`bit`	PostgreSQL přijímá `true`/`false`; Microsoft SQL používá `1`/`0`.
`BYTEA`	`varbinary(max)`	Stejný koncept, jiné jméno.
`JSONB`	`nvarchar(max)` s funkcemi JSON	Microsoft SQL ukládá JSON jako text a ověřuje pomocí `ISJSON()`. Viz data JSON.
`TIMESTAMP WITH TIME ZONE`	`datetimeoffset`	Oba ukládají offset. Viz Datetime zpracování.
`INTERVAL`	Žádný přímý ekvivalent	Vypočítejte s `DATEADD()` a `DATEDIFF()`.
`ARRAY`	Žádný přímý ekvivalent	Použijte samostatnou tabulku, JSON pole nebo `STRING_SPLIT()`.
`UUID`	`uniqueidentifier`	Ovladač `mssql-python` nativně mapuje `uuid.UUID`. Viz konfigurace modulu.
`NOW()` / `CURRENT_TIMESTAMP`	`GETDATE()` nebo `SYSDATETIME()`	`SYSDATETIME()` Dává vyšší přesnost.
`LIMIT 10 OFFSET 20`	`OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY`	Vyžaduje klauzuli `ORDER BY` .
`\\|\\|` (strunová konkace)	`+` nebo `CONCAT()`	`CONCAT()` zpracovává `NULL` hodnoty.
`COALESCE(a, b)`	`COALESCE(a, b)` nebo `ISNULL(a, b)`	`COALESCE` je identický v obou případech.
`string_agg(col, ',')`	`STRING_AGG(col, ',')`	Dostupné v SQL Server 2017+.
`RETURNING id`	`OUTPUT INSERTED.id`	Použijte `OUTPUT` ve větě `INSERT`, `UPDATE`, nebo `DELETE` .
`ON CONFLICT ... DO UPDATE`	Prohlášení `MERGE`	`MERGE` podporuje INSERT + UPDATE + DELETE v jednom příkazu. Viz vzory přepisu dotazů.
`EXPLAIN ANALYZE`	`SET STATISTICS IO ON; SET STATISTICS TIME ON;`	Nebo použijte plány provedení v SSMS / Azure Data Studio.
`\d tablename`	`sp_help 'tablename'`	Nebo zadejte dotaz `INFORMATION_SCHEMA.COLUMNS`.
`pg_dump`	`bcp`, `BACKUP DATABASE`	Použití `bulkcopy()` pro programové načítání dat z Python.

CREATE TABLE Příklad

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

Vzory přepisování dotazů

Následující sekce ukazují běžné vzory dotazů PostgreSQL a jejich ekvivalenty v T-SQL.

Pagination

PostgreSQL:

cursor.execute("SELECT * FROM products ORDER BY name LIMIT %s OFFSET %s", (10, 20))

mssql-python:

cursor.execute(
    "SELECT * FROM Production.Product ORDER BY Name OFFSET ? ROWS FETCH NEXT ? ROWS ONLY",
    (20, 10)
)

Pořadí parametrů je obrácené. Microsoft SQL umisťuje OFFSET před FETCH NEXT.

Upsert (vložit nebo aktualizovat záznam)

PostgreSQL zpracovává ON CONFLICT insert-or-update na základě jediného omezení:

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

Microsoft SQL zpracovává MERGEINSERT, UPDATE, a DELETE v jednom příkazu. Použijte klauzuli USING s aliasy parametrů:

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

Pro hromadné upserty uspořádejte řádky do dočasné tabulky pomocí bulkcopy(), a pak MERGE z ní. Další informace naleznete v části Hromadný upsert s přechodnou tabulkou.

Vložte 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 pracuje s příkazy INSERT, UPDATE a DELETE. Může vracet více sloupců.

Značky parametrů

psycopg2 používá %s pro polohové parametry a %(name)s pro pojmenované parametry. Ovladač mssql-python používá ? pro poziční a %(name)s pro pojmenované:

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

Rozdíly mezi transakcemi a autocommitem

PostgreSQL (psycopg2) automaticky otevře transakci na první příkaz a vyžaduje explicitní commit():

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

Ovladač mssql-python funguje ve výchozím nastavení stejně. Autocommit je vypnutý a vy výslovně voláte commit() :

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

Chcete-li povolit automatické potvrzování:

PsycopG2:

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

mssql-python:

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

Viz Správa transakcí, kde najdete informace o úrovních izolace, bodech obnovení a vzorech opakování při uváznutí.

Typové úvahy

Následující sekce pokrývají nejčastější rozdíly v mapování typů mezi PostgreSQL a Microsoft SQL.

JSON

PostgreSQL má nativní JSONB s indexačními a dotazovými operátory (->, ->>, ). @> Microsoft SQL ukládá JSON jako nvarchar(max) a poskytuje funkce pro dotazování:

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

V jazyce Python oba přístupy používají k serializaci json.dumps():

import json

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

Viz data JSON, kde najdete úplné pokyny k vzorům ukládání a dotazování na data JSON.

Univerzálně jedinečný identifikátor (UUID)

Jak PostgreSQL, tak mssql-python nativně mapují uuid.UUID:

import uuid

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

Viz Konfigurace modulu pro native_uuid možnost připojení.

Datum a časové pásmo

TIMESTAMPTZ se v PostgreSQL při uložení převádí na UTC. Microsoft SQL zachovává datetimeoffset původní 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})

Pokud potřebujete konzistentní UTC úložiště, převeďte v Python před vložením:

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

Úplné mapování typů naleznete v části Práce s datem a časem.

Pole

PostgreSQL podporuje nativní sloupce typu pole (INTEGER[], TEXT[]). Microsoft SQL nemá typ pole. Běžné alternativy:

Samostatná tabulka (normalizovaná). Nejlepší pro dotazovatelná, indexovaná data.
JSON pole uložené v nvarchar(max). Dobré pro neprůhledná metadata.
Řetězec oddělený čárkami s STRING_SPLIT(). Jednoduché, ale omezené.

# 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 ve výchozím nastavení ukládá veškerý text jako UTF-8. Microsoft SQL rozlišuje mezi varcharem (kódování kódové stránky) a nvarcharem (UTF-16). Ovladač mssql-python ve výchozím nastavení posílá Python str hodnoty jako nvarchar, takže Unicode text funguje bez další konfigurace. Pokud vaše schéma používá varcharovy sloupce a potřebujete se vyhnout implicitní konverzi, použijte setinputsizes() k určení typu sloupce. Podrobnosti o kódování naleznete v sekci String a Unicode data .

Hromadné načítání a přesun dat

PostgreSQL používá COPY pro hromadné operace. MSSQL-Python poskytuje 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)

U velkých souborů použijte generátor, abyste se vyhnuli načítání celého souboru do paměti:

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)

Viz Hromadné kopírovací operace pro mapování sloupců, zpracování identity a tipy na výkon.

Migrace schémat a dat

Použijte tento přístup k migraci stávající databáze PostgreSQL:

Exportujte schéma. Použijte pg_dump --schema-only k získání DDL. Pro podrobnosti o možnostech a okrajových případech (vlastnictví, oprávnění, rozšíření a filtrování) viz odkaz PostgreSQLpg_dump. Přepiš DDL pomocí tabulky rozdílů SQL dialektů .
Vytvářejte tabulky v Microsoft SQL. Spusť přepsaný DDL proti cílové databázi.
Export dat. Použijte pg_dump --data-only --format=csv, nebo se pomocí psycopg2 dotazujte na každou tabulku. Pro velké datové sady a kompatibilitní spínače si prostudujte dokumentaci PostgreSQLpg_dump, zejména sekci možností.
Načíst data pomocí Bulk Copy. Přečti si pořadí sloupců z katalogu, abys nemusel tvrdě kódovat seznam sloupců v každé tabulce, a pak každou tabulku streamovat do Microsoft SQL. Tady je ukázkový skript:

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

Ve výchozím nastavení tento skript migruje každou tabulku, která existuje jak v PostgreSQL, tak public v dbo (SQL Server), seřazenou podle závislostí na cizích klíčích. Nastavte TABLE_MAPPINGS na explicitní seznam, pokud chcete migrovat pouze podmnožinu.

To předpokládá, že zdroj a cíl používají stejné názvy sloupců, což je obvyklý případ po přepisu DDL. Pomocník automaticky zpracovává sloupec identity: keep_identity uchovává zdrojové primární klíče, když má cíl IDENTITY sloupec, takže odkazy na cizí klíče zůstávají nedotčené. Aby SQL Server místo toho přiřadil nové klíče, vylučte sloupec identity z columns a předejte keep_identity=False.

Cizí klíče a omezení

bulkcopy() používá protokol TDS bulk insert, který během načítání nevynucuje cizí klíče ani nekontroluje omezení. Bez explicitního požadavku na jejich kontrolu SQL Server ignoruje omezení CHECK během hromadného importu a následně je označí FOREIGN KEY jako nedůvěryhodné, jak je popsáno v .BULK INSERT Toto chování má pro migraci dva praktické důsledky:

Na pořadí načítání nezáleží. Podřízenou tabulku můžete načíst dříve než její nadřazenou tabulku, aniž by došlo k porušení omezení cizího klíče. Zachovejte primární klíče pomocí keep_identity=True, stejně jako to dělá pomocná rutina, aby se hodnoty klíčů nadřazeného a podřízeného záznamu i po načtení stále shodovaly.
Omezení se nakonec považují za nedůvěryhodná. Po hromadném načtení je každý cizí klíč označen jako nedůvěryhodný (sys.foreign_keys.is_not_trusted = 1), protože jej SQL Server neověřil. Poslední krok ve skriptu znovu ověřuje každou načtenou tabulku pomocí ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Tento krok označuje důvěryhodná omezení, aby je optimalizátor dotazů mohl využít, a odhaluje špatná data. Pokud podřízený řádek odkazuje na chybějícího rodiče, příkaz selže kvůli porušení integrity omezení, které omezení pojmenuje, takže můžete opuštěné řádky opravit ještě před spustěním.

Limitations

Před migrací si tyto rozdíly prostudujte:

Téma	PostgreSQL	mssql-python / SQL Server
`callproc()`	Podporováno	Zvyšuje `NotSupportedError`. Místo toho použijte `cursor.execute("EXECUTE ...")`.
Parametry s hodnotou typu tabulka (TVP)	Žádný přímý ekvivalent	Není podporováno v aktuálním ovladači. Používejte temp tables nebo JSON pro víceřádkové parametry.
Nativní `ARRAY` sloupce	Podporováno	Žádný typ pole. Použijte normalizované tabulky, JSON pole nebo `STRING_SPLIT()`.
`LISTEN`/`NOTIFY`	Podporováno	Žádný přímý ekvivalent. Použijte Service Broker nebo dotazování na úrovni aplikace.
`COPY` Streamování	Podporováno	Použití `bulkcopy()` pro hromadné načítání dat.
Vrácení upravených řádků	klauzule `RETURNING`	`OUTPUT INSERTED` / `OUTPUT DELETED` klauzule v příkazech DML.
Asynchronní ovladač	`psycopg3` má nativní asynchronní podporu	`mssql-python` Podpora asynchronních zařízení je orientována na obcházení (pool vláken).
Fulltextové vyhledávání	`tsvector` / `tsquery`	`CONTAINS()` / `FREETEXT()` s fulltextovými indexy.
ORM (SQLAlchemy)	Plně podporovaná	Podporováno vestavěným dialektem mssql-python v SQLAlchemy 2.1.0b2+ (před vydáním).

Kontrolní seznam pro ověření

Použijte tento kontrolní seznam k ověření vaší migrace:

Všechny značky parametrů %s nahraďte parametry ? nebo %(name)s.
Ujistěte se, že všechny %(name)s parametry stále fungují (oba ovladače tento formát podporují).
Přepis LIMIT/OFFSET na .OFFSET/FETCH NEXT
Přepis RETURNING na OUTPUT INSERTED.
Přepis ON CONFLICT na MERGE.
Nahraďte SERIAL / BIGSERIAL .IDENTITY
BOOLEAN sloupce nahrazeny bitem.
Nahraďte sloupce pole normalizovanými tabulkami nebo JSONem.
Nahraďte operátory JSONB pomocí JSON_VALUE() / JSON_QUERY().
Aktualizujte připojovací řetězec pro Microsoft SQL autentizaci.
Vyzkoušejte aplikaci proti AdventureWorks nebo vašemu cílovému schématu.

Autentizace a nasazení

Samostatně spravované aplikace PostgreSQL se obvykle nasazují s připojovacími řetězci, které obsahují hesla, nebo používají soubory .pgpass a proměnné prostředí PGPASSWORD. Azure Database for PostgreSQL podporuje autentizaci Microsoft Entra, takže pokud už používáte autentizaci bez hesla, stejný model identity se přenáší i do Azure SQL.

Pro produkční workloady proti Azure SQL použijte managed identity:

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

Pro lokální vývoj a CI viz Container a lokální vývoj pro vzory nastavení pipeline Docker, devcontainer a CI.

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-07-22

Migrate from PostgreSQL to Microsoft SQL with mssql-python

Co získáte přechodem na Microsoft SQL

Instalace ovladače

Vytvoření databáze SQL

Aktualizujte kód připojení

Připojovací řetězce

Autentizace

Kontextoví manažeři

Sdílení připojení

Rozdíly v SQL dialektech

CREATE TABLE Příklad

Vzory přepisování dotazů

Pagination

Upsert (vložit nebo aktualizovat záznam)

Vložte ID

Značky parametrů

Rozdíly mezi transakcemi a autocommitem

Typové úvahy

JSON

Univerzálně jedinečný identifikátor (UUID)

Datum a časové pásmo

Pole

Unicode

Hromadné načítání a přesun dat

Migrace schémat a dat

Cizí klíče a omezení

Limitations

Kontrolní seznam pro ověření

Autentizace a nasazení

Související obsah

Váš názor

Další materiály