Migracja z PostgreSQL do Microsoft SQL za pomocą mssql-python

Wiele zespołów Python najpierw uczy się PostgreSQL. Gdy Twoje obciążenie potrzebuje funkcji takich jak tabele czasowe, pełna MERGE semantyka czy indeksy columnstore, przejdź do Microsoft SQL. Ten przewodnik obejmuje główne decyzje i zmiany w kodzie przy przenoszeniu aplikacji Python z PostgreSQL (za pomocą psycopg2 lub psycopg3) do Microsoft SQL za pomocą sterownikamssql-python.

Note

Jeśli migrujesz z Azure Database for PostgreSQL, obie usługi obsługują uwierzytelnianie Microsoft Entra oraz zarządzaną tożsamość. Zmiany w kodzie zawarte w tym przewodniku mają zastosowanie niezależnie od tego, czy Twoje źródło PostgreSQL jest samodzielnie zarządzane, czy hostowane w Azure.

Co zyskujesz, przechodząc do Microsoft SQL

Microsoft SQL zawiera funkcje upraszczające bezpieczeństwo, zgodność i operacje dla obciążeń produkcyjnych. Zrozum te funkcje przed rozpoczęciem migracji, aby móc z nich skorzystać podczas przejścia:

Dynamiczne maskowanie danych i bezpieczeństwo na poziomie wiersza. Maskuj kolumny dla użytkowników, którzy nie potrzebują pełnego dostępu, i ograniczaj widoczność wierszy przez politykę bezpieczeństwa. Te funkcje działają z każdym kierowcą.
Tabele temporalne (wersjonowane systemowo). Microsoft SQL automatycznie śledzi historię wiersza. Brak wyzwalaczy, brak tabel audytu, brak kodu aplikacji.
Pełna semantyka MERGE. Pojedyncza instrukcja obsługuje INSERT, UPDATE i DELETE za pomocą klauzuli OUTPUT na potrzeby śladów audytowych. Klauzula ON CONFLICT w PostgreSQL dotyczy wyłącznie operacji wstawienia lub aktualizacji dla jednego ograniczenia.
Indeksy Columnstore. Dodaj kolumnową pamięć do istniejących tabel dla hybrydowych obciążeń OLTP/analityki. Nie potrzebna jest osobna baza danych analitycznych.
Uwierzytelnianie identyfikatora Entra firmy Microsoft. Połącz się z zarządzanymi tożsamościami, podmiotami usługowymi lub interaktywnym logowaniem. Azure Database for PostgreSQL obsługuje także uwierzytelnianie Microsoft Entra, więc jeśli już z niego korzystasz, przejście jest proste.

Instalowanie sterownika

Zanim zaczniesz, upewnij się, że masz Python 3.10 lub nowszy oraz docelową bazę danych SQL.

Tworzenie bazy danych SQL

Stwórz lub połącz się z bazą danych SQL na jednej z następujących platform:

Tworzenie bazy danych SQL w ciągu kilku minut przy użyciu witryny Azure Portal

Sterowniki PostgreSQL wymagają zewnętrznych natywnych bibliotek.

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

Sterownik mssql-python łączy swoją natywną warstwę. Na Windows nie potrzebujesz zewnętrznego menedżera sterowników ani pakietów systemowych.

pip install mssql-python

Na Linuksie i macOS instaluj niewielki zestaw bibliotek systemowych udokumentowanych w Instalacji. Nie ma odpowiednika dla pg_config ani libpq-dev.

Zaktualizuj kod połączenia

W poniższych sekcjach omówiono najważniejsze zmiany dotyczące parametrów połączenia, uwierzytelniania, menedżerów kontekstu oraz pul połączeń.

Łańcuchy połączenia

psycopg2 używa łańcucha znaków DSN lub argumentów słów kluczowych.

import psycopg2

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

mssql-python obsługuje także argumenty słów kluczowych, co pozwala uniknąć problemów z kodowaniem URL, które często pojawiają się w ciągu połączeń SQLAlchemy, gdy hasła zawierają @, ;, lub {} znaki.

import mssql_python

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

Lub użyj parametrów połączenia.

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

Pełny zestaw słów kluczowych parametry połączenia można znaleźć w artykule Connection strings.

Authentication

Uwierzytelnianie PostgreSQL zazwyczaj wykorzystuje pg_hba.conf reguły z nazwą użytkownika i hasłem. Azure Database for PostgreSQL obsługuje również uwierzytelnianie Microsoft Entra. Microsoft SQL obsługuje wiele trybów uwierzytelniania za pomocą jednego słowa kluczowego połączenia:

Podejście PostgreSQL	Odpowiednik mssql-python
Nazwa użytkownika i hasło	`UID=...;PWD=...;`
Szyfrowanie SSL/TLS	`Encrypt=yes;`(domyślnie włączony dla Azure SQL)
Entra auth (Azure PostgreSQL)	`Authentication=ActiveDirectoryDefault;` (bez hasła)
Managed identity (Azure PostgreSQL)	`Authentication=ActiveDirectoryMSI;`
Service principal (Azure PostgreSQL)	`Authentication=ActiveDirectoryServicePrincipal;`

Użyj ActiveDirectoryDefault do lokalnego programowania. Automatycznie łączy się z Azure CLI, zmiennymi środowiskowymi oraz zarządzaną tożsamością. W środowisku produkcyjnym użyj określonego trybu, takiego jak ActiveDirectoryMSI (tożsamość zarządzana) lub ActiveDirectoryServicePrincipal, aby uniknąć powolnego przeszukiwania łańcucha poświadczeń. Zobacz uwierzytelnianie Microsoft Entra, aby uzyskać informacje o wszystkich siedmiu trybach uwierzytelniania.

Menedżerowie kontekstu

Oba sterowniki obsługują menedżerów kontekstu, ale zachowanie różni się:

psycopg2 with conn: zatwierdza transakcję po pomyślnym zakończeniu i wycofuje ją w przypadku wyjątku, ale nie zamyka połączenia:

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

Element with conn: w mssql-python zamyka połączenie po zakończeniu. Prace niezaangażowane są cofnięte:

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

Buforowanie połączeń

psycopg2 wymaga wyraźnego ustawienia i zarządzania pulą połączeń.

from psycopg2 import pool

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

mssql-python sterownik ma domyślnie włączoną wbudowaną funkcję puli połączeń. Nie jest potrzebna żadna konfiguracja.

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

Skonfiguruj rozmiar puli, jeśli domyślne nie pasują do twojego obciążenia.

import mssql_python

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

Aby uzyskać wskazówki dotyczące wielkości basenu i rozwiązywania problemów z wyczerpaniem wody, zobacz Pool Connection.

Różnice w dialekcie SQL

Poniższa tabela mapuje typowe wzorce PostgreSQL na ich odpowiedniki Transact-SQL (T-SQL):

PostgreSQL	SQL Server (T-SQL)	Notatki
`SERIAL` / `BIGSERIAL`	`int IDENTITY(1,1)`	Microsoft SQL używa `IDENTITY` do automatycznego zwiększania.
`TEXT`	`nvarchar(max)`	Zastosowanie `nvarchar` w Unicode. Użyj `nvarchar(4000)` lub krótszej wersji, jeśli dane na to pozwalają.
`BOOLEAN`	`bit`	PostgreSQL akceptuje `true`/`false`; Microsoft SQL używa `1`/`0`.
`BYTEA`	`varbinary(max)`	Ten sam koncept, inna nazwa.
`JSONB`	`nvarchar(max)` z funkcjami JSON	Microsoft SQL przechowuje JSON jako tekst i waliduje za pomocą `ISJSON()`. Zobacz dane JSON.
`TIMESTAMP WITH TIME ZONE`	`datetimeoffset`	Oba składują offset. Zobacz Obsługa daty i godziny.
`INTERVAL`	Brak bezpośredniego odpowiednika	Oblicz za pomocą `DATEADD()` i `DATEDIFF()`.
`ARRAY`	Brak bezpośredniego odpowiednika	Użyj osobnej tabeli, tablicy JSON lub `STRING_SPLIT()`.
`UUID`	`uniqueidentifier`	Sterownik `mssql-python` natywnie obsługuje `uuid.UUID`. Zobacz Konfiguracja modułu.
`NOW()` / `CURRENT_TIMESTAMP`	`GETDATE()` lub `SYSDATETIME()`	`SYSDATETIME()` Daje większą precyzję.
`LIMIT 10 OFFSET 20`	`OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY`	Wymaga klauzuli `ORDER BY` .
`\\|\\|` (struna konkata)	`+` lub `CONCAT()`	`CONCAT()` obsługuje `NULL` wartości.
`COALESCE(a, b)`	`COALESCE(a, b)` lub `ISNULL(a, b)`	`COALESCE` jest taki sam w obu przypadkach.
`string_agg(col, ',')`	`STRING_AGG(col, ',')`	Dostępny w SQL Server 2017+.
`RETURNING id`	`OUTPUT INSERTED.id`	Użycie `OUTPUT` w zdaniu `INSERT`, `UPDATE`, lub `DELETE` .
`ON CONFLICT ... DO UPDATE`	`MERGE`, oświadczenie	`MERGE` obsługuje INSERT + UPDATE + DELETE w jednej instrukcji. Zobacz wzorce przepisywania zapytań.
`EXPLAIN ANALYZE`	`SET STATISTICS IO ON; SET STATISTICS TIME ON;`	Albo użyj planów wykonawczych w SSMS / Azure Data Studio.
`\d tablename`	`sp_help 'tablename'`	Lub wyślij zapytanie `INFORMATION_SCHEMA.COLUMNS`.
`pg_dump`	`bcp`, `BACKUP DATABASE`	Użyj `bulkcopy()` do programowego ładowania danych z poziomu Pythona.

CREATE TABLE Przykład

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

Wzorce przepisywania zapytań

Poniższe sekcje pokazują typowe wzorce zapytań PostgreSQL oraz ich odpowiedniki 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)
)

Kolejność parametrów jest odwrócona. Microsoft SQL stawia OFFSET przed FETCH NEXT.

Upsert (wstaw lub zaktualizuj)

Funkcja PostgreSQL ON CONFLICT obsługuje wstawianie lub aktualizację dla jednego ograniczenia:

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

Microsoft SQL MERGE obsługuje INSERT, UPDATE, i DELETE w jednym zdaniu. Użyj klauzuli USING z aliasami parametrów:

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

W przypadku zbiorczych operacji upsert umieść wiersze w tabeli tymczasowej za pomocą bulkcopy(), a następnie wykonaj MERGE z tej tabeli. Więcej informacji można znaleźć w artykule Bulk upsert with a staging table.

Pobierz wstawione 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 Działa z INSERT, UPDATE, i DELETE zdaniami. Może zwracać wiele kolumn.

Znaczniki parametrów

psycopg2 używa %s parametrów pozycyjnych oraz %(name)s parametrów nazwanych. Sterownik mssql-python używa ? dla parametrów pozycyjnych i %(name)s dla parametrów nazwanych:

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

Różnice między transakcjami a automatycznymi zatwierdzeniami

PostgreSQL (psycopg2) automatycznie otwiera transakcję przy pierwszym poleceniu i wymaga jawnego commit():

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

Sterownik mssql-python działa domyślnie tak samo. Autocommit jest wyłączony i jawnie wywołujesz commit():

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

Aby włączyć automatyczne zatwierdzanie:

PsycopG2:

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

mssql-python:

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

Zobacz zarządzanie transakcjami , aby poznać poziomy izolacji, punkty zapisu i wzorce ponownych prób zablokowania.

Rozważania dotyczące typu

Poniższe sekcje omawiają najczęstsze różnice w mapowaniu typów między PostgreSQL a Microsoft SQL.

JSON

PostgreSQL ma natywną obsługę JSONB z indeksowaniem i operatorami zapytań (->, ->>, @>). Microsoft SQL przechowuje JSON jako nvarchar(max) i udostępnia funkcje do zapytań:

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

W Pythonie oba podejścia używają json.dumps() do serializacji:

import json

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

Zobacz dane JSON, aby uzyskać pełne wskazówki dotyczące przechowywania danych JSON i wzorców zapytań.

UUID

Zarówno PostgreSQL, jak i mssql-python obsługują uuid.UUID natywnie:

import uuid

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

Zobacz Konfigurację modułu dla opcji połączenia native_uuid .

Czas i strefa czasowa

TIMESTAMPTZ w PostgreSQL jest przekształcany na UTC podczas zapisu. Element datetimeoffset w Microsoft SQL zachowuje oryginalny 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})

Jeśli potrzebujesz stałego przechowywania UTC, konwertuj w Python przed wstawieniem:

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

Zobacz obsługę czasu datowego dla pełnego mapowania typów.

Tablice

PostgreSQL obsługuje natywne kolumny tablic (INTEGER[], TEXT[]). Microsoft SQL nie ma typu tablicy. Popularne alternatywy:

Oddzielna tabela (znormalizowana). Najlepsze do danych, które można indeksować i przeszukiwać za pomocą zapytań.
Matryca JSON przechowywana w nvarchar(max). Dobre do nieprzejrzystych metadanych.
Ciąg rozdzielony komami z STRING_SPLIT(). Proste, ale ograniczone.

# 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 domyślnie przechowuje cały tekst jako UTF-8. Microsoft SQL rozróżnia varchar (kodowanie stron kodowych) i nvarchar (UTF-16). Sterownik mssql-python domyślnie wysyła wartości Python str jako nvarchar, więc tekst Unicode działa bez dodatkowej konfiguracji. Jeśli Twój schemat używa kolumn varchar i musisz uniknąć konwersji domyślnej, użyj setinputsizes() do określenia typu kolumny. Zobacz Dane typu String i Unicode, aby uzyskać szczegółowe informacje o kodowaniu.

Ładowanie masowe i przenoszenie danych

PostgreSQL używa COPY do operacji zbiorczych. mssql-python oferuje 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)

W przypadku dużych plików użyj generatora, aby uniknąć ładowania całego pliku do pamięci:

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)

Zobacz operacje kopiowania masowego , aby uzyskać wskazówki dotyczące mapowania kolumn, obsługi tożsamości i wydajności.

Migracja schematów i danych

Skorzystaj z takiego podejścia do migracji istniejącej bazy danych PostgreSQL:

Eksportuj schemat. Użyj pg_dump --schema-only, aby uzyskać DDL. Szczegóły opcji i przypadki brzegowe (własność, uprawnienia, rozszerzenia i filtrowanie) można znaleźć w referencji PostgreSQLpg_dump. Przepisz DDL za pomocą tabeli różnic dialektów SQL .
Twórz tabele w Microsoft SQL. Uruchom przepisany DDL na docelowej bazie danych.
Eksportowanie danych. Użyj pg_dump --data-only --format=csv lub wykonaj zapytanie do każdej tabeli za pomocą psycopg2. W przypadku dużych zbiorów danych i przełączników kompatybilności, zapoznaj się z dokumentacją PostgreSQLpg_dump, szczególnie sekcją opcji.
Załaduj dane za pomocą kopiowania zbiorczego. Przeczytaj kolejność kolumn docelowych z katalogu, żeby nie zakodować na stałe listy kolumn w każdej tabeli, a potem przesyłać każdą tabelę do Microsoft SQL. Oto przykładowy skrypt:

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

Domyślnie ten skrypt migruje każdą tabelę istniejącą zarówno w public (PostgreSQL), jak i dbo (SQL Server), uporządkowaną według zależności klucza obcego. Ustaw TABLE_MAPPINGS na jawną listę, jeśli chcesz zmigrować tylko podzbiór elementów.

Zakłada się, że źródło i miejsce docelowe używają tych samych nazw kolumn, co zwykle ma miejsce po przepisaniu DDL. Pomocnik automatycznie obsługuje kolumnę tożsamości: keep_identity zachowuje źródłowe klucze podstawowe, gdy tabela docelowa ma kolumnę IDENTITY, dzięki czemu referencje kluczy obcych pozostają nienaruszone. Aby zamiast tego umożliwić programowi SQL Server przypisanie nowych kluczy, wyklucz kolumnę identyfikacyjną z columns i przekaż keep_identity=False.

Klucze obce i ograniczenia

bulkcopy() używa protokołu TDS bulk insert, który nie wymusza obcego klucza ani nie sprawdza ograniczeń podczas ładowania. Bez wyraźnego żądania sprawdzenia ich SQL Server ignoruje ograniczenia CHECK i FOREIGN KEY podczas importu zbiorczego i oznacza je później jako niezaufane, jak opisano w BULK INSERT. Takie zachowanie ma dwie praktyczne konsekwencje dla migracji:

Kolejność ładowania nie ma znaczenia. Możesz załadować tabelę potomną przed jej rodzicem, nie natrafiając na naruszenia klucza obcego. Zachowaj klucze główne za pomocą keep_identity=True, tak jak robi to funkcja pomocnicza, aby wartości kluczy nadrzędnych i podrzędnych nadal się zgadzały po załadowaniu.
Ograniczenia stają się niezaufane. Po załadowaniu zbiorczym każdy klucz obcy jest oznaczony jako nieufny (sys.foreign_keys.is_not_trusted = 1), ponieważ SQL Server go nie zweryfikował. Ostatni krok w skrypcie ponownie waliduje każdą załadowaną tabelę za pomocą ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Ten krok oznacza ograniczenia jako zaufane, dzięki czemu optymalizator zapytań może je wykorzystać, a także wykrywa nieprawidłowe dane. Jeśli wiersz podrzędny odwołuje się do brakującego wiersza nadrzędnego, instrukcja kończy się błędem naruszenia ograniczenia integralności, wskazującym nazwę ograniczenia, dzięki czemu możesz naprawić osierocone wiersze przed uruchomieniem produkcyjnym.

Limitations

Przejrzyj te różnice przed migracją:

Temat	PostgreSQL	mssql-python / SQL Server
`callproc()`	Wsparte	Podbija.`NotSupportedError` Użyj `cursor.execute("EXECUTE ...")` zamiast tego.
Parametry o wartości tabelarycznej (TVP)	Brak bezpośredniego odpowiednika	Nie jest obsługiwane w bieżącym sterowniku. Używaj tabel tymczasowych lub formatu JSON do przekazywania parametrów wielowierszowych.
Kolumny natywne `ARRAY`	Wsparte	Brak typu tablicy. Używaj tabel znormalizowanych, tablic JSON lub `STRING_SPLIT()`.
`LISTEN`/`NOTIFY`	Wsparte	Brak bezpośredniego odpowiednika. Użyj Service Brokera lub ankiet na poziomie aplikacji.
`COPY` Streaming	Wsparte	Użyj `bulkcopy()` do masowego ładowania danych.
Zwracanie zmodyfikowanych wierszy	klauzula `RETURNING`	`OUTPUT INSERTED` / `OUTPUT DELETED` klauzula w instrukcjach DML.
Sterownik asynchroniczny	`psycopg3` ma natywną asynchroniczność	`mssql-python` Wsparcie asynchroniczne jest nastawione na obejścia (pula wątków).
Wyszukiwanie pełnotekstowe	`tsvector` / `tsquery`	`CONTAINS()` / `FREETEXT()` z indeksami pełnotekstowymi.
ORM (SQLAlchemy)	W pełni wspierane	Obsługiwane przez wbudowany dialekt mssql-python w SQLAlchemy 2.1.0b2+ (wersja przedpremierowa).

Lista kontrolna weryfikacji poprawności

Użyj tej listy kontrolnej, aby zweryfikować swoją migrację:

Zamień wszystkie %s markery parametrów na ? lub %(name)s parametry.
Upewnij się, że wszystkie %(name)s parametry nadal działają (oba sterowniki obsługują ten format).
Przekształc LIMIT/OFFSET do .OFFSET/FETCH NEXT
Przekształc RETURNING do OUTPUT INSERTED.
Przekształc ON CONFLICT do MERGE.
Zastąp SERIAL / BIGSERIAL na .IDENTITY
BOOLEAN kolumny zastąpione bitem.
Zamień kolumny tablic na znormalizowane tabele lub JSON.
Zastąp JSONB operatory naJSON_VALUE() / JSON_QUERY() .
Zaktualizuj parametry połączenia dla uwierzytelniania Microsoft SQL.
Testuj aplikację względem AdventureWorks lub swojego docelowego schematu.

Uwierzytelnianie i wdrożenie

Samozarządzające aplikacje PostgreSQL zazwyczaj wdrażają się z łańcuchami połączeń zawierającymi hasła lub korzystają z .pgpass plików i PGPASSWORD zmiennych środowiskowych. Azure Database for PostgreSQL obsługuje uwierzytelnianie Microsoft Entra, więc jeśli już używasz uwierzytelniania bez hasła, ten sam model tożsamości przenosi się na Azure SQL.

Dla obciążeń produkcyjnych przeciwko Azure SQL użyj managed identity:

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

W przypadku lokalnego programowania i CI zobacz Kontenery i programowanie lokalne, aby uzyskać informacje o wzorcach konfiguracji dla Dockera, devcontainera i potoków CI.

Opinia

Czy ta strona była pomocna?

Last updated on 2026-07-22

Migracja z PostgreSQL do Microsoft SQL za pomocą mssql-python

Co zyskujesz, przechodząc do Microsoft SQL

Instalowanie sterownika

Tworzenie bazy danych SQL

Zaktualizuj kod połączenia

Łańcuchy połączenia

Authentication

Menedżerowie kontekstu

Buforowanie połączeń

Różnice w dialekcie SQL

CREATE TABLE Przykład

Wzorce przepisywania zapytań

Pagination

Upsert (wstaw lub zaktualizuj)

Pobierz wstawione ID

Znaczniki parametrów

Różnice między transakcjami a automatycznymi zatwierdzeniami

Rozważania dotyczące typu

JSON

UUID

Czas i strefa czasowa

Tablice

Unicode

Ładowanie masowe i przenoszenie danych

Migracja schematów i danych

Klucze obce i ograniczenia

Limitations

Lista kontrolna weryfikacji poprawności

Uwierzytelnianie i wdrożenie

Treści powiązane

Opinia

Dodatkowe zasoby