Migrer de PostgreSQL vers Microsoft SQL avec mssql-python

De nombreuses équipes Python apprennent d’abord PostgreSQL. Lorsque votre charge de travail a besoin de fonctionnalités comme des tables temporelles, une sémantique complète MERGE ou des index de colonne, migrez vers Microsoft SQL. Ce guide couvre les principales décisions et modifications de code pour transférer une application Python de PostgreSQL (utilisant psycopg2 ou psycopg3) vers Microsoft SQL en utilisant le mssql-python pilote.

Ce que vous gagnez en passant à Microsoft SQL

Microsoft SQL inclut des fonctionnalités qui simplifient la sécurité, la conformité et les opérations pour les charges de travail en production. Comprenez ces fonctionnalités avant de commencer la migration afin de pouvoir en profiter pendant la transition :

• Masquage dynamique des données et sécurité au niveau des lignes. Masquez les colonnes pour les utilisateurs qui n’ont pas besoin d’un accès complet, et restreignez la visibilité des lignes en fonction de la politique de sécurité. Ces fonctionnalités fonctionnent avec n’importe quel pilote.
• Tables temporelles (version système). Microsoft SQL suit automatiquement l’historique des lignes. Pas de déclencheurs, pas de tables d’audit, pas de code applicatif.
• Sémantique complète MERGE. Une seule instruction gère INSERT, UPDATE, et DELETE avec une clause OUTPUT pour les traces d’audit. La clause PostgreSQL ON CONFLICT ne couvre que l’insertion ou la mise à jour sur une seule contrainte.
• Index columnstore. Ajoutez du stockage en colonnes aux tables existantes pour des charges de travail hybrides OLTP/analytique. Aucune base de données analytique distincte n’est nécessaire.
• Microsoft Entra ID authentification. Connectez-vous avec des identités gérées, des responsables de service ou une connexion interactive. Azure Database pour PostgreSQL prend également en charge l'authentification Microsoft Entra, donc si vous l'utilisez déjà, la transition est simple.

Installer le pilote

Avant de commencer, assurez-vous d’avoir Python 3.10 ou plus récent ainsi qu’une base de données SQL cible.

Créer une base de données SQL

Créer ou connecter une base de données SQL sur l’une des plateformes suivantes :

Les pilotes PostgreSQL nécessitent des bibliothèques natives externes.

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

Le mssql-python pilote regroupe sa couche native. Sous Windows, vous n'avez pas besoin d'un gestionnaire de pilotes externe ni de paquets système.

pip install mssql-python

Sous Linux et macOS, installez un petit ensemble de bibliothèques système documentées dans Installation. Il n’existe pas d’équivalent à pg_config ou libpq-dev.

Mettre à jour le code de connexion

Les sections suivantes couvrent les changements clés des chaînes de connexion, de l’authentification, des gestionnaires de contexte et du pooling.

Chaînes de connexion

psycopg2 utilise une chaîne DSN ou des arguments de mots-clés.

import psycopg2

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

MSSQL-Python prend également en charge les arguments de mots-clés, ce qui évite les problèmes de codage d’URL que les chaînes de connexion SQLAlchemy rencontrent souvent lorsque les mots de passe contiennent @, ;, ou {} des caractères.

import mssql_python

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

Ou utilisez une chaîne de connexion.

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

Pour l’ensemble complet des mots-clés de chaîne de connexion, voir Chaînes de connexion.

Authentication

L’authentification PostgreSQL utilise généralement des règles pg_hba.conf basées sur un nom d’utilisateur et un mot de passe. Azure Database pour PostgreSQL prend également en charge l’authentification Microsoft Entra. Microsoft SQL prend en charge plusieurs modes d’authentification via un seul mot-clé de connexion :

Utiliser ActiveDirectoryDefault pour le développement local. Il enchaîne automatiquement avec Azure CLI, les variables d’environnement et l’identité gérée. En production, utilisez un mode spécifique comme ActiveDirectoryMSI (identité gérée) ou ActiveDirectoryServicePrincipal pour éviter le lent parcours de la chaîne des informations d’identification. Voir l’authentification Microsoft Entra pour les sept modes d’authentification.

Gestionnaires de contexte

Les deux pilotes supportent les gestionnaires de contexte, mais le comportement diffère :

psycopg2 with conn: valide la transaction en cas de succès et l’annule en cas d’exception, mais ne ferme pas la connexion :

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: ferme la connexion à la sortie. Le travail non engagé est annulé :

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

Regroupement de connexions

PsyCopg2 nécessite une mise en place et une gestion explicites d’un pool de connexions.

from psycopg2 import pool

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

Le mssql-python pilote a activé par défaut le pooling intégré. Aucune installation n’est nécessaire.

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

Configurez la taille du pool si les paramètres par défaut ne correspondent pas à votre charge de travail.

import mssql_python

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

Pour obtenir des conseils sur le dimensionnement du pool et la résolution des problèmes d’épuisement du pool, consultez Pool de connexions.

Différences entre les dialectes SQL

Le tableau suivant associe les modèles PostgreSQL courants à leurs équivalents Transact-SQL (T-SQL) :

CREATE TABLE Exemple

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

Motifs de réécriture de requêtes

Les sections suivantes présentent les schémas de requête PostgreSQL courants et leurs équivalents 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)
)

L’ordre des paramètres est inversé. Microsoft SQL met OFFSET avant FETCH NEXT.

Upsert (insérer ou mettre à jour)

PostgreSQL ON CONFLICT gère l’insertion ou la mise à jour sur une seule contrainte :

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

Microsoft SQL MERGE gère INSERT, UPDATE, et DELETE en une seule phrase. Utilisez une USING clause avec des alias de paramètres :

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

Pour les opérations d’upsert en masse, placez les lignes dans une table temporaire à l’aide de bulkcopy(), puis utilisez MERGE à partir de celle-ci. Pour plus d’informations, voir Bulk upsert avec une table de stastation.

Faites entrer votre identifiant

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 fonctionne avec les instructions INSERT, UPDATE et DELETE. Il peut renvoyer plusieurs colonnes.

Marqueurs de paramètres

Psycopg2 utilise %s les paramètres de position et %(name)s les paramètres nommés. Le pilote mssql-python utilise ? pour les paramètres positionnels et %(name)s pour les paramètres nommé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}
)

Différences entre transaction et validation automatique

PostgreSQL (psycopg2) ouvre automatiquement une transaction dès la première commande et requiert un commit() explicite :

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

Le mssql-python pilote fonctionne de la même manière par défaut. L’autocommit est désactivé, et vous appelez commit() explicitement :

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

Pour activer l’autocommit :

Psycopg2 :

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

MSSQL-python :

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

Voir la gestion des transactions pour les niveaux d’isolement, les points de sauvegarde et les schémas de réévaluation des blocages.

Considérations de type

Les sections suivantes couvrent les différences de mappage de types les plus courantes entre PostgreSQL et Microsoft SQL.

JSON

PostgreSQL dispose d’opérateurs natifs JSONB d’indexation et de requête (->, ->>, @>). Microsoft SQL stocke JSON sous la forme nvarchar(max) et fournit des fonctions pour les requêtes :

En Python, les deux approches utilisent json.dumps() pour la sérialisation :

import json

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

Voir les données JSON pour des conseils complets sur le stockage JSON et les motifs d’interrogation.

Identifiant Unique Universel (UUID)

PostgreSQL et mssql-python se mappent uuid.UUID nativement :

import uuid

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

Voir Configuration du module pour l’option native_uuid connexion.

Date, heure et fuseau horaire

Le TIMESTAMPTZ de PostgreSQL est converti en UTC lors du stockage. Microsoft SQL datetimeoffset conserve le décalage original :

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

Si vous avez besoin d’un stockage UTC cohérent, convertissez en Python avant d’insérer :

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

Voir Gestion des dates et heures pour la correspondance complète des types.

Tableaux

PostgreSQL prend en charge les colonnes natives de tableau (INTEGER[], TEXT[]). Microsoft SQL n'a pas de type de tableau. Alternatives courantes :

1. Table séparée (normalisée). Idéal pour des données indexées et interrogables.
2. Tableau JSON stocké dans nvarchar(max). Bon pour les métadonnées opaques.
3. Chaîne séparée par virgules avec STRING_SPLIT(). Simple mais limité.

# 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 stocke tout le texte sous forme d’UTF-8 par défaut. Microsoft SQL distingue varchar (encodage de page de code) et nvarchar (UTF-16). Le mssql-python pilote envoie par défaut des valeurs Python str sous forme de nvarchar, donc le texte Unicode fonctionne sans configuration supplémentaire. Si votre schéma utilise des colonnes varchar et que vous devez éviter la conversion implicite, utilisez setinputsizes() pour spécifier le type de colonne. Voir Données de chaîne et Unicode pour les détails d’encodage.

Chargement en masse et déplacement des données

PostgreSQL utilise COPY pour les opérations groupées. MSSQL-Python fournit 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)

Pour les fichiers volumineux, utilisez un générateur pour éviter de charger l’intégralité du fichier en mémoire :

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)

Voir les opérations de copie en bloc pour les correspondances de colonnes, la gestion des identités et des conseils sur les performances.

Migration des schémas et des données

Utilisez cette approche pour migrer une base de données PostgreSQL existante :

1. Exportez le schéma. Utilisez pg_dump --schema-only pour obtenir le DDL. Pour les détails des options et les cas particuliers (propriété, privilèges, extensions et filtrage), voir la référence PostgreSQLpg_dump. Réécrivez le DDL en utilisant la table des différences dialectales SQL .
2. Créer des tables dans Microsoft SQL. Exécutez le DDL réécrit sur votre base de données cible.
3. Exporter des données. Utilisez pg_dump --data-only --format=csv ou interrogez chaque table avec psycopg2. Pour les grands ensembles de données et les switches de compatibilité, consultez la documentation PostgreSQLpg_dump, en particulier la section options.
4. Chargez des données avec bulkcopy. Lisez l'ordre des colonnes de destination dans le catalogue pour ne pas coder en dur une liste de colonnes par tableau, puis intégrez chaque table dans Microsoft SQL. Voici un exemple de script :

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

Par défaut, ce script migre toutes les tables existantes à la fois public dans (PostgreSQL) et dbo (SQL Server), ordonnées par dépendances à clé étrangère. Définissez TABLE_MAPPINGS sur une liste explicite si vous souhaitez ne migrer qu’un sous-ensemble.

Cela suppose que la source et la destination utilisent les mêmes noms de colonnes, ce qui est le cas habituel après avoir réécrit le DDL. L’assistant gère automatiquement la colonne d’identité : keep_identity préserve les clés primaires source lorsque la destination a une IDENTITY colonne, de sorte que les références aux clés étrangères restent intactes. Pour permettre à SQL Server d’assigner de nouvelles clés à la place, excluez la colonne identité de columns et passez keep_identity=False.

Clés étrangères et contraintes

bulkcopy() utilise le protocole TDS bulk insert, qui n’impose pas de clés étrangères ni de contraintes de vérification pendant le chargement. En l’absence de demande explicite de vérification, SQL Server ignore les contraintes CHECK et FOREIGN KEY lors d’une importation en bloc et les marque ensuite comme non fiables, comme décrit dans BULK INSERT. Ce comportement a deux conséquences pratiques pour la migration :

• L’ordre de chargement n’a pas d’importance. Vous pouvez charger une table enfant avant sa mère sans toucher à des violations de clé étrangère. Conservez les clés primaires avec keep_identity=True, comme le fait l’assistant, afin que les valeurs des clés parent et enfant correspondent toujours après le chargement.
• Les contraintes finissent par ne pas être dignes de confiance. Après un chargement en masse, chaque clé étrangère est marquée comme non fiable (sys.foreign_keys.is_not_trusted = 1) car SQL Server ne l'a pas vérifiée. La dernière étape du script revalide chaque table chargée avec ALTER TABLE ... WITH CHECK CHECK CONSTRAINT ALL. Cette étape marque les contraintes fiables pour que l’optimiseur de requêtes puisse les utiliser, et met en lumière des données défaillantes. Si une ligne enfant contient une référence à une ligne parente absente, l’instruction échoue avec une violation de contrainte d’intégrité indiquant le nom de la contrainte, ce qui vous permet de corriger les lignes orphelines avant la mise en production.

Limitations

Examinez ces différences avant de migrer :

Liste de contrôle de validation

Utilisez cette liste de contrôle pour vérifier votre migration :

1. Remplacez tous les marqueurs de paramètre %s par des paramètres ? ou %(name)s.
2. Assurez-vous que tous les %(name)s paramètres fonctionnent toujours (les deux pilotes prennent en charge ce format).
3. Réécrire LIMIT/OFFSET en .OFFSET/FETCH NEXT
4. Réécrire RETURNING en OUTPUT INSERTED.
5. Réécrire ON CONFLICT en MERGE.
6. Remplacez SERIAL / BIGSERIAL par IDENTITY.
7. BOOLEAN colonnes remplacées par bit.
8. Remplacez les colonnes de tableau par des tables normalisées ou du JSON.
9. Remplacez les opérateurs JSONB par JSON_VALUE() / JSON_QUERY().
10. Mettre à jour la chaîne de connexion pour l’authentification SQL Microsoft.
11. Testez l’application contre AdventureWorks ou votre schéma cible.

Authentification et déploiement

Les applications PostgreSQL autogérées se déploient généralement avec des chaînes de connexion contenant des mots de passe, ou utilisent .pgpass des fichiers et PGPASSWORD des variables d’environnement. Azure Database pour PostgreSQL prend en charge l'authentification Microsoft Entra, donc si vous utilisez déjà une authentification sans mot de passe, le même modèle d'identité s'applique à Azure SQL.

Pour les charges de travail de production sur Azure SQL, utilisez l’identité managée :

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

Pour le développement local et le CI, voir Conteneur et développement local pour les modèles de configuration Docker, devcontainer et pipeline CI.

Contenu connexe

• Installation
• Chaînes de connexion
• Authentification Microsoft Entra
• Opérations de copie en bloc
• Mappages de types de données
• Gestion des transactions
• Développement dans des conteneurs et en local
• Migrer depuis SQLite