mssql-pythonを使ってPostgreSQLからMicrosoft SQLへ移行

多くのPythonチームはまずPostgreSQLを学びます。 時間テーブルやフルMERGEセマンティクス、カラムストアインデックスなどの機能が必要なら、Microsoft SQLに移行してください。 本ガイドでは、PythonアプリケーションをPostgreSQL(psycopg2またはpsycopg3使用)からmssql-pythonドライバを使ってMicrosoft SQLに移行するための主な決定とコード変更について説明します。

Microsoft SQLに移行することで得られるもの

Microsoft SQLには、本番ワークロードのセキュリティ、コンプライアンス、運用を簡素化する機能が含まれています。 移行を始める前にこれらの機能を理解し、移行時に活用できるようにしましょう:

• 動的データマスキング と 行レベルのセキュリティ。 完全なアクセスを必要としないユーザー向けにマスキングカラムを設置し、セキュリティポリシーで行の可視性を制限します。 これらの機能はどのドライバーにも対応可能です。
• 時空テーブル (システムバージョン付き)。 Microsoft SQLは行履歴を自動的に追跡します。 トリガーも監査テーブルもアプリケーションコードもありません。
• 完全なセマンティクス MERGE 。 単一の文で INSERT、 UPDATE、 DELETE を処理し、監査トレイル用のOUTPUT節があります。 PostgreSQLの ON CONFLICT 節は、単一の制約に対する挿入または更新のみを対象としています。
• コラムストアのインデックス。 既存のテーブルに列状ストレージを追加し、ハイブリッドなOLTP/アナリティクスワークロードを活用しましょう。 別途分析データベースは必要ありません。
• Microsoft Entra ID 認証。 マネージドID、サービスプリンシパル、またはインタラクティブサインインと接続できます。 Azure Database for PostgreSQLはMicrosoft Entra認証もサポートしているので、すでに使っているなら移行は簡単です。

ドライバーをインストールする

始める前に、Python 3.10以降のソフトとターゲットとなるSQLデータベースを持っていることを確認してください。

SQL データベースを作成する

以下のいずれかのプラットフォームでSQLデータベースを作成または接続してください:

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はキーワード引数もサポートしており、パスワードに @文字、 ;文字、または {} 文字が含まれている場合にSQLAlchemy接続文字列でよくあるURLエンコーディングの問題を回避します。

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 Database for PostgreSQLはMicrosoft Entra認証もサポートしています。 Microsoft SQLは、単一の接続キーワードを通じて複数の認証モードをサポートしています:

ローカル開発には ActiveDirectoryDefault を使用します。 Azure CLI、環境変数、マネージドIDを自動でチェーン化します。 本番環境では、 ActiveDirectoryMSI (マネージドアイデンティティ)や ActiveDirectoryServicePrincipal のような特定のモードを使い、遅い認証情報のチェーンウォークを避けましょう。 7つの認証モードすべてについては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)

プールのサイズ設定やプール枯渇のトラブルシューティングについては、「 接続プーリング」をご覧ください。

SQL方言の違い

以下の表は、一般的なPostgreSQLパターンをそれらの Transact-SQL(T-SQL)対応にマッピングしています:

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対応物を示します。

ページネーション

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を1つの文で処理します。 パラメータエイリアス付きの 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))

バルクアップサートの場合は、 bulkcopy()を使って一時テーブルの行をステージングし、そこから MERGE します。 詳細は「 ステージングテーブル付きのバルクアップサート」をご覧ください。

挿入された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)として保存し、クエリのための関数を提供します:

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

完全な型マッピングについては Datetimeの処理 を参照してください。

配列

PostgreSQLはネイティブの配列列(INTEGER[]、 TEXT[])をサポートしています。 Microsoft SQLには配列型がありません。 一般的な代替案:

1. 別の表 (正規化済み)。 クエリ可能でインデックス化されたデータに最適です。
2. nvarchar(max)に保存されたJSON配列。 不透明なメタデータには良いです。
3. コンマ区切りの文字列、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ドライバーはデフォルトでstrとしてPython 送信するため、Unicodeのテキストは追加の設定なしで動作します。 スキーマが varchar 列を使っていて暗黙の変換を避けたい場合は、 setinputsizes() でカラムタイプを指定します。 符号化の詳細については「 文字列および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データベースを移行するには、この方法を用いてください:

1. スキーマをエクスポートします。 pg_dump --schema-onlyを使ってDDLを取得しましょう。 オプションの詳細や例外ケース(所有権、権限、拡張機能、フィルタリング)については、PostgreSQLのpg_dump参照を参照してください。 SQL方言の違い表を使ってDDLを書き直します。
2. Microsoft SQLでテーブルを作成します。 書き直したDDLをターゲットデータベースに対して実行してください。
3. データをエクスポートします。 pg_dump --data-only --format=csvを使うか、各テーブルにpsycopg2でクエリを送ります。 大規模なデータセットや互換性切り替えについては、PostgreSQLの pg_dump ドキュメント、特にオプションセクションを確認してください。
4. データを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 から ID 列を除外し、keep_identity=False を渡します。

外部鍵と制約

bulkcopy() TDSのバルクインサートプロトコルを使用しており、これはロード中に外部キーやチェック制約を強制しません。 明示的にチェックを求める要求がなければ、SQL ServerはCHECKおよびFOREIGN KEY制約を一括インポート中に無視し、その後に信頼できないとマークします(BULK INSERTで説明されています)。 この行動は移住に2つの実用的な影響をもたらします。

• 読み込み順は問いません。 子テーブルは、親テーブルより先に読み込んでも、外部キー制約違反は発生しません。 ヘルパーと同様に、主キーは keep_identity=Trueで保持し、ロード後も親キーと子キーの値が一致します。
• 制約は信頼できなくなります。 一括読み込み後、SQL Serverで検証されていないため、各外部キーは信頼されていない (sys.foreign_keys.is_not_trusted = 1) としてマークされます。 スクリプトの最終ステップでは、 ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALLで読み込まれたすべてのテーブルを再検証します。 このステップは、クエリオプティマイザが利用できるように制約を信頼しているものを示し、悪いデータを表面化します。 子行が存在しない親行を参照している場合、そのステートメントは制約名が示された整合性制約違反となって失敗するため、本番稼働前に孤児行を修正しておくことができます。

Limitations

移行前にこれらの違いを確認してください:

検証チェックリスト

このチェックリストを使って移行を確認します:

1. すべての %s パラメータマーカーを ? または %(name)s パラメータに置き換えます。
2. すべての %(name)s パラメータが動作していることを確認してください(両方のドライバーがこのフォーマットをサポートしています)。
3. LIMIT / OFFSETをOFFSET/FETCH NEXTに書き直してください。
4. RETURNINGをOUTPUT INSERTEDに書き直してください。
5. ON CONFLICTをMERGEに書き直してください。
6. SERIAL / BIGSERIALをIDENTITYに置き換えてください。
7. BOOLEAN 柱は ビットに置き換えられています。
8. 配列列を正規化されたテーブルやJSONに置き換えましょう。
9. オペレーターJSONBJSON_VALUE() / JSON_QUERY()に置き換えましょう。
10. Microsoft SQL認証用に接続文字列を更新してください。
11. AdventureWorksやターゲットスキーマに対してアプリケーションをテストしてください。

認証と展開

自己管理型のPostgreSQLアプリケーションは通常、パスワードを含む接続文字列で展開するか、 .pgpass ファイルと PGPASSWORD 環境変数を使用します。 Azure Database for PostgreSQLはMicrosoft Entra認証をサポートしているので、すでにパスワードレス認証を使っている場合、同じIDモデルがAzure SQLにも引き継がれます。

Azure SQL に対する本番ワークロードには、managed identity を使いましょう:

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

ローカル開発およびCIについては、Docker、devcontainer、CIパイプラインの設定パターンについては 「コンテナおよびローカル開発 」を参照してください。

関連するコンテンツ

• Installation
• 接続文字列
• Microsoft Entra 認証
• 一括コピー作業
• データ型のマッピング
• トランザクション管理
• コンテナーとローカル開発
• SQLiteからの移行