Переход с PostgreSQL на Microsoft SQL с помощью mssql-python

Многие команды Python сначала изучают PostgreSQL. Когда вашей рабочей нагрузке нужны такие функции, как временные таблицы, полная MERGE семантика или индексы columnstore, переходите на Microsoft SQL. В этом руководстве рассматриваются основные решения и изменения кода для переноса Python-приложения с PostgreSQL (с использованием psycopg2 или psycopg3) на Microsoft SQL с помощью драйвераmssql-python.

Замечание

Если вы переходите из База данных Azure для PostgreSQL, оба сервиса поддерживают аутентификацию Microsoft Entra и управляемую идентификацию. Изменения в коде, описанные в этом руководстве, применимы независимо от того, управляется ли исходный экземпляр PostgreSQL самостоятельно или размещён в Azure.

Что вы получаете, переходя на Microsoft SQL

Microsoft SQL включает возможности, упрощающие безопасность, соответствие требованиям и эксплуатацию производственных нагрузок. Изучите эти функции до начала миграции, чтобы воспользоваться ими во время перехода:

Динамическое маскирование данных и безопасность на уровне строк. Маскируйте столбцы для пользователей, которым не нужен полный доступ, и ограничивайте видимость строк по политике безопасности. Эти функции работают с любым водителем.
Временные таблицы (системные версии). Microsoft SQL автоматически отслеживает историю строк. Нет триггеров, нет таблиц аудита, нет кода приложения.
Полная MERGE семантика. Один оператор обрабатывает INSERT, UPDATE и DELETE с предложением OUTPUT для ведения журнала аудита. Клауза PostgreSQL ON CONFLICT охватывает только вставку или обновление по одному ограничению.
Индексы Columnstore. Добавьте колонковое хранилище в существующие таблицы для гибридных OLTP/аналитических нагрузок. Отдельная аналитическая база данных не нужна.
Проверка подлинности идентификатора Microsoft Entra. Подключайтесь к управляемым идентичностям, принципалам сервиса или интерактивным входом. База данных Azure для PostgreSQL также поддерживает аутентификацию Microsoft Entra, поэтому, если вы уже используете её, переход будет простым.

Установка драйвера

Прежде чем начать, убедитесь, что у вас есть Python 3.10 или выше и целевая база данных SQL.

Создание базы данных SQL

Создайте или подключитесь к SQL-базе данных на одной из следующих платформ:

Создание базы данных SQL в минутах с помощью портала Azure

Драйверы PostgreSQL требуют внешних нативных библиотек.

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

Драйвер mssql-python включает свой нативный слой. В Windows не нужен внешний менеджер драйверов или системные пакеты.

pip install mssql-python

На Linux и macOS установите небольшой набор системных библиотек, задокументированных в Инсталляции. Нет эквивалента ни pg_config, ни libpq-dev.

Обновить код соединения

Следующие разделы охватывают изменения ключей в строках соединения, аутентификацию, менеджеры контекста и пулирование.

Строки подключения

psycopg2 использует аргументы DSN-строк или ключевых слов.

import psycopg2

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

MSSQL-Python также поддерживает аргументы ключевых слов, что позволяет избежать проблем с кодированием URL, которые часто возникают в строках соединения SQLAlchemy, когда пароли содержат @, ;, или {} символы.

import mssql_python

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

Или используйте строка подключения.

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

Полный набор ключевых слов строки подключения см. в разделе Connection strings.

Authentication

Аутентификация PostgreSQL обычно использует pg_hba.conf правила с именем пользователя и паролем. База данных Azure для PostgreSQL также поддерживает аутентификацию Microsoft Entra. Microsoft SQL поддерживает несколько режимов аутентификации с помощью одного ключевого слова соединения:

Подход PostgreSQL	Эквивалент mssql-python
Имя пользователя и пароль	`UID=...;PWD=...;`
SSL/TLS-шифрование	`Encrypt=yes;`(включено по умолчанию для Azure SQL)
Entra auth (Azure PostgreSQL)	`Authentication=ActiveDirectoryDefault;` (без пароля)
Managed identity (Azure PostgreSQL)	`Authentication=ActiveDirectoryMSI;`
Service principal (Azure PostgreSQL)	`Authentication=ActiveDirectoryServicePrincipal;`

Используется ActiveDirectoryDefault для локальной разработки. Он автоматически интегрируется через Azure CLI, переменные среды и управляемую идентичность. Для продакшена используйте конкретный режим, например ActiveDirectoryMSI (управляемая идентичность) или ActiveDirectoryServicePrincipal чтобы избежать медленного перехода по цепочке учетных данных. См. Аутентификация Microsoft Entra, где описаны все семь режимов аутентификации.

Менеджеры контекста

Оба драйвера поддерживают менеджеры контекста, но поведение отличается:

Psycopg2 with conn: фиксирует транзакцию при успешном выполнении и выполняет откат при исключении, но не закрывает соединение:

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 with conn: закрывает соединение при выходе. Незавершённая работа откатывается:

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

Пулинг соединений

Psycopg2 требует явной настройки и управления пулом соединений.

from psycopg2 import pool

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

В драйвере mssql-python встроенное объединение подключений включено по умолчанию. Настройка не требуется.

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

Настройте размер пула, если настройки по умолчанию не подходят под вашу нагрузку.

import mssql_python

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

Для рекомендаций по размеру бассейна и устранению проблем с истощением бассейна см. раздел Connection pooling.

Различия в диалектах SQL

Следующая таблица сопоставляет распространённые шаблоны PostgreSQL с их Transact-SQL (T-SQL) эквивалентами:

PostgreSQL	SQL Server (T-SQL)	Примечания.
`SERIAL` / `BIGSERIAL`	`int IDENTITY(1,1)`	Microsoft SQL использует `IDENTITY` для автоинкремента.
`TEXT`	`nvarchar(max)`	Используйте `nvarchar` для Unicode. Предпочитайте `nvarchar(4000)` или сокращайте время, когда данные позволяют.
`BOOLEAN`	`bit`	PostgreSQL принимает `true`/`false`; Microsoft SQL использует `1`/`0`.
`BYTEA`	`varbinary(max)`	Та же концепция, но другое название.
`JSONB`	`nvarchar(max)` с JSON-функциями	Microsoft SQL хранит JSON в виде текста и проверяет его с помощью `ISJSON()`. См. данные JSON.
`TIMESTAMP WITH TIME ZONE`	`datetimeoffset`	Оба сохраняют смещение. См. обработку даты и времени.
`INTERVAL`	Нет прямого эквивалента	Вычислите с `DATEADD()` и `DATEDIFF()`.
`ARRAY`	Нет прямого эквивалента	Используйте отдельную таблицу, массив JSON или `STRING_SPLIT()`.
`UUID`	`uniqueidentifier`	Драйвер `mssql-python` изначально сопоставляет `uuid.UUID`. См. конфигурацию модуля.
`NOW()` / `CURRENT_TIMESTAMP`	`GETDATE()` или `SYSDATETIME()`	`SYSDATETIME()` даёт большую точность.
`LIMIT 10 OFFSET 20`	`OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY`	Требуется условие `ORDER BY`.
`\\|\\|` (конкатенация строк)	`+` или `CONCAT()`	`CONCAT()` Обрабатывает `NULL` значения.
`COALESCE(a, b)`	`COALESCE(a, b)` или `ISNULL(a, b)`	`COALESCE` идентичен в обоих случаях.
`string_agg(col, ',')`	`STRING_AGG(col, ',')`	Доступно в SQL Server 2017+.
`RETURNING id`	`OUTPUT INSERTED.id`	Используйте `OUTPUT` в инструкции `INSERT`, `UPDATE` или `DELETE`.
`ON CONFLICT ... DO UPDATE`	заявление `MERGE`	`MERGE` поддерживает INSERT + UPDATE + DELETE в одном операторе. См. Шаблоны переписывания запросов.
`EXPLAIN ANALYZE`	`SET STATISTICS IO ON; SET STATISTICS TIME ON;`	Или используйте планы выполнения в SSMS / Azure Data Studio.
`\d tablename`	`sp_help 'tablename'`	Или выполните запрос `INFORMATION_SCHEMA.COLUMNS`.
`pg_dump`	`bcp`, `BACKUP DATABASE`	Используйте `bulkcopy()` для программной загрузки данных из Python.

CREATE TABLE пример

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

Шаблоны переписывания запросов

В следующих разделах представлены распространённые шаблоны запросов PostgreSQL и их 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)
)

Порядок параметров обратный. Microsoft SQL ставит OFFSET перед FETCH NEXT.

Upsert (вставить или обновить)

PostgreSQL ON CONFLICT обрабатывает вставку или обновление на одном ограничении:

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

Microsoft SQL MERGE обрабатывает INSERT, UPDATE и DELETE в одном операторе. Используйте USING клаузу с псевдонимами параметров:

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

Для пакетных операций upsert поместите строки во временную таблицу с помощью bulkcopy(), а затем выполните MERGE из неё. Для получения дополнительной информации см. раздел Bulk upsert с таблицей сцены.

Вставьте 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 работает с операторами INSERT, UPDATE и DELETE. Он может возвращать несколько столбцов.

Маркеры параметров

PsyCOPG2 использует %s позиционные параметры и %(name)s именованные параметры. Драйвер mssql-python использует ? для позиционных параметров и %(name)s для именованных:

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

Различия между транзакциями и автоподтверждением

PostgreSQL (psycopg2) автоматически открывает транзакцию при первой команде и требует явного commit():

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

mssql-python Драйвер работает так же по умолчанию. Автокоммит отключён, и вы явно вызываете commit() :

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

Для включения автокоммита:

Psycopg2:

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

mssql-python:

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

См. «Управление транзакциями» для получения сведений об уровнях изоляции, точках сохранения и шаблонах повторных попыток при взаимоблокировке.

Типовые аспекты

В следующих разделах рассматриваются наиболее распространённые различия в отображении типов между PostgreSQL и Microsoft SQL.

JSON

PostgreSQL имеет нативные JSONB операторы индексации и запросов (->, ->>, @>). Microsoft SQL хранит JSON как nvarchar(max) и предоставляет функции для запросов:

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

В Python оба подхода используют 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"})}
)

Смотрите данные JSON для полных рекомендаций по шаблонам хранения и запросов JSON.

UUID (Универсальный уникальный идентификатор)

И PostgreSQL, и mssql-python нативно сопоставляют uuid.UUID:

import uuid

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

См. раздел «Конфигурация модуля » для опции native_uuid соединения.

Дата, время и часовой пояс

В PostgreSQL TIMESTAMPTZ преобразуется в UTC при сохранении. Microsoft SQL datetimeoffset сохраняет исходное смещение:

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

Если вам нужно стабильное UTC-хранилище, конвертируйте в Python перед вставкой:

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

См. Обработка даты и времени для полного сопоставления типов.

Массивы

PostgreSQL поддерживает нативные столбцы массивов (INTEGER[], TEXT[]). В Microsoft SQL нет типа массива. Распространённые альтернативы:

Отдельная таблица (нормализованная). Лучше всего подходит для данных, поддерживающих запросы и индексирование.
JSON-массив хранится в nvarchar(max). Хорошо для непрозрачных метаданных.
Строка, разделённая запятыми с STRING_SPLIT(). Просто, но ограниченно.

# 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 по умолчанию хранит весь текст как UTF-8. Microsoft SQL различает varchar (кодирование страниц) и nvarchar (UTF-16). mssql-python Драйвер по умолчанию отправляет значения Python как str, поэтому текст Unicode работает без дополнительной настройки. Если ваша схема использует столбцы варчара и вам нужно избежать неявного преобразования, используйте setinputsizes() для указания типа столбца. См. данные String и Unicode для подробностей кодирования.

Массовая загрузка и перемещение данных

PostgreSQL использует COPY для массовых операций. MSSQL-Python предоставляет 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)

Для больших файлов используйте генератор, чтобы избежать загрузки всего файла в память:

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)

См. раздел «Операции массового копирования » для отображения столбцов, обработки идентичности и советов по производительности.

Схема и миграция данных

Используйте этот подход для миграции существующей базы данных PostgreSQL:

Экспортируйте схему. Используйте pg_dump --schema-only, чтобы получить DDL. Для подробностей опций и крайних случаев (владение, привилегии, расширения и фильтрация) см. ссылку на PostgreSQLpg_dump. Перепишите DDL с помощью таблицы различий SQL-диалектов .
Создавайте таблицы в Microsoft SQL. Проверьте переписанный DDL с вашей целевой базой данных.
Экспорт данных. Используйте pg_dump --data-only --format=csv или задавайте запросы к каждой таблице с помощью psycopg2. Для больших наборов данных и совместимых коммутаторов ознакомьтесь с документацией PostgreSQLpg_dump, особенно разделом с опциями.
Загрузите данные с помощью утилиты bulkcopy. Читайте порядок столбцов назначения из каталога, чтобы не создавать жёсткий код для каждой таблицы, а затем транслировать каждую таблицу в Microsoft SQL. Ниже приведен пример сценария:

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

По умолчанию этот скрипт мигрирует все таблицы, которые существуют и в public (PostgreSQL), и в dbo (SQL Server), в порядке зависимостей по внешним ключам. Установите TABLE_MAPPINGS в явный список, если хотите мигрировать только часть.

Это предполагает, что исходник и цель используют одни и те же имена столбцов, что обычно происходит после переписывания DDL. Вспомогательный компонент автоматически обрабатывает столбец идентификаторов: keep_identity сохраняет исходные первичные ключи, когда в целевой таблице есть столбец IDENTITY, поэтому ссылки по внешним ключам сохраняются. Чтобы SQL Server мог назначать новые ключи, исключите столбец идентичности из columns и передайте keep_identity=False.

Внешние ключи и ограничения

bulkcopy() использует протокол TDS bulk insert, который не применяет ограничения внешнего ключа или проверки во время загрузки. Если явно не запрашивать проверку ограничений CHECK и FOREIGN KEY, SQL Server игнорирует их во время операции массового импорта, а затем помечает как недоверенные, как описано в BULK INSERT. Такое поведение имеет два практических последствия для миграции:

Порядок загрузки не имеет значения. Вы можете загрузить дочернюю таблицу раньше родительской, не сталкиваясь с нарушениями внешних ключей. Сохраняйте первичные ключи с keep_identity=True, как это делает помощник, чтобы значения родительского и дочернего ключей оставались совпадением после загрузки.
Ограничения в итоге становятся недоверенными. После массовой загрузки каждый внешний ключ отмечается как недоверенный (sys.foreign_keys.is_not_trusted = 1), потому что SQL Server не проверил его. Последний шаг скрипта повторно проверяет каждую загруженную таблицу с ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Этот шаг отмечает ограничения, которым доверяют, чтобы оптимизатор запросов мог их использовать, и приводит к выводу некачественных данных. Если дочерняя строка ссылается на несуществующую родительскую строку, оператор завершается с ошибкой из-за нарушения ограничения целостности с указанием имени ограничения, так что вы можете исправить строки-сироты до ввода системы в эксплуатацию.

Limitations

Изучите эти различия перед миграцией:

Тема	PostgreSQL	mssql-python / SQL Server
`callproc()`	Поддерживается	Увеличивает `NotSupportedError`. Вместо этого используйте `cursor.execute("EXECUTE ...")`.
Параметры, представленные в виде таблицы (TVP)	Нет прямого эквивалента	В текущем драйвере не поддерживается. Используйте временные таблицы или JSON для многостроковых параметров.
Встроенные `ARRAY` столбцы	Поддерживается	Нет типа массива. Используйте нормализованные таблицы, JSON-массивы или `STRING_SPLIT()`.
`LISTEN`/`NOTIFY`	Поддерживается	Нет прямого эквивалента. Используйте Service Broker или опросы на уровне приложения.
`COPY` Стриминг	Поддерживается	Используйте `bulkcopy()` для массовой загрузки данных.
Возвращение модифицированных строк	Клаузула `RETURNING`	`OUTPUT INSERTED` / `OUTPUT DELETED` предложение в операторах DML.
Асинхронный драйвер	`psycopg3` имеет нативный асинхрон	`mssql-python` Поддержка async ориентирована на обходные решения (пул потоков).
Полнотекстовый поиск	`tsvector` / `tsquery`	`CONTAINS()` / `FREETEXT()` с полнотекстовыми индексами.
ORM (SQLAlchemy)	полностью поддерживается.	Поддерживается через встроенный диалект mssql-python в SQLAlchemy 2.1.0b2+ (до релиза).

Контрольный список проверки

Используйте этот чек-лист для подтверждения вашей миграции:

Замените все маркеры параметров %s на параметры ? или %(name)s.
Убедитесь, что все %(name)s параметры работают (оба драйвера поддерживают этот формат).
Перепишите LIMIT/OFFSET в OFFSET/FETCH NEXT.
Перепишите RETURNING в OUTPUT INSERTED.
Перепишите ON CONFLICT в MERGE.
Заменить SERIAL / BIGSERIAL на .IDENTITY
BOOLEAN столбцы заменены на бит.
Замените столбцы массива нормализованными таблицами или JSON.
Заменить JSONB операторы на JSON_VALUE() / JSON_QUERY().
Обновить строку подключения для аутентификации Microsoft SQL.
Тестируйте приложение с использованием AdventureWorks или целевой схемы.

Аутентификация и развертывание

Самоуправляемые приложения PostgreSQL обычно развёртываются с помощью строк соединения с паролями или используют .pgpass файлы и PGPASSWORD переменные среды. База данных Azure для PostgreSQL поддерживает аутентификацию Microsoft Entra, поэтому если вы уже используете аутентификацию без пароля, та же модель идентичности переносится и в Azure SQL.

Для производственных нагрузок против Azure SQL используйте управляемую идентичность:

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

Для локальной разработки и CI см. раздел Container and local development для шаблонов настройки Docker, devcontainer и CI pipeline.

Обратная связь

Были ли сведения на этой странице полезными?

Last updated on 2026-07-22

Переход с PostgreSQL на Microsoft SQL с помощью mssql-python

Что вы получаете, переходя на Microsoft SQL

Установка драйвера

Создание базы данных SQL

Обновить код соединения

Строки подключения

Authentication

Менеджеры контекста

Пулинг соединений

Различия в диалектах SQL

CREATE TABLE пример

Шаблоны переписывания запросов

Pagination

Upsert (вставить или обновить)

Вставьте ID

Маркеры параметров

Различия между транзакциями и автоподтверждением

Типовые аспекты

JSON

UUID (Универсальный уникальный идентификатор)

Дата, время и часовой пояс

Массивы

Unicode

Массовая загрузка и перемещение данных

Схема и миграция данных

Внешние ключи и ограничения

Limitations

Контрольный список проверки

Аутентификация и развертывание

Связанные материалы

Обратная связь

Дополнительные ресурсы