Exécuter des requêtes avec mssql-python

Le pilote mssql-python fournit des méthodes de curseur pour l’exécution de requêtes SQL, les requêtes paramétrées, les opérations batch et les instructions préparées.

Exécution de requête de base

Utilisez la méthode d’un execute() curseur pour exécuter des instructions SQL :

import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("SELECT Name, ListPrice FROM Production.Product WHERE Color = 'Black'")
rows = cursor.fetchall()

for row in rows:
    print(row.Name, row.ListPrice)

cursor.close()
conn.close()

Requêtes paramétrables

Utilisez toujours des requêtes paramétrables pour empêcher l’injection SQL. Le paramstyle par défaut du pilote est pyformat (paramètres fictifs nommés), mais il prend aussi en charge qmark (paramètres fictifs positionnels). Utilisez qmark pour les séquences d’échappement ODBC {CALL}.

Style pyformat (par défaut)

Utilisez des espaces réservés nommés avec la syntaxe %(name)s et transmettez un dictionnaire :

cursor.execute(
    "SELECT Name, ListPrice FROM Production.Product WHERE Color = %(color)s AND ListPrice > %(price)s",
    {"color": "Black", "price": 10.00}
)

style Qmark

Utilisez des placeholders positionnels avec ? et passez un tuple ou une liste :

cursor.execute(
    "SELECT Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = ? AND ListPrice > ?",
    (1, 10.00)
)

Le pilote détecte automatiquement le style de paramètres en fonction de votre requête SQL et de vos types de paramètres.

opérations INSERT, UPDATE, DELETE

Pour les instructions de modification de données, utilisez des requêtes paramétrées et commencez la transaction :

cursor.execute("CREATE TABLE #ExecDemo (Name NVARCHAR(50), CategoryID INT, Price DECIMAL(10,2))")
cursor.execute(
    "INSERT INTO #ExecDemo (Name, CategoryID, Price) VALUES (%(name)s, %(category)s, %(price)s)",
    {"name": "New Product", "category": 1, "price": 19.99}
)
conn.commit()

print(f"Rows affected: {cursor.rowcount}")

Exécution par lots avec executemany()

Utilisez executemany() pour insérer efficacement plusieurs lignes. Le pilote utilise la liaison de paramètres par colonne pour des performances optimales :

products = [
    {"name": "Product A", "category": 1, "price": 10.00},
    {"name": "Product B", "category": 1, "price": 15.00},
    {"name": "Product C", "category": 2, "price": 20.00},
]

cursor.execute("CREATE TABLE #BatchDemo (Name NVARCHAR(50), CategoryID INT, Price DECIMAL(10,2))")
cursor.executemany(
    "INSERT INTO #BatchDemo (Name, CategoryID, Price) VALUES (%(name)s, %(category)s, %(price)s)",
    products
)
conn.commit()

print(f"Rows inserted: {cursor.rowcount}")

Avec le style qmark :

products = [
    ("Product A", 1, 10.00),
    ("Product B", 1, 15.00),
    ("Product C", 2, 20.00),
]

cursor.execute("CREATE TABLE #QmarkDemo (Name NVARCHAR(50), CategoryID INT, Price DECIMAL(10,2))")
cursor.executemany(
    "INSERT INTO #QmarkDemo (Name, CategoryID, Price) VALUES (?, ?, ?)",
    products
)
conn.commit()

Exécution par lot de plusieurs instructions

Utilisez batch_execute() sur la connexion pour exécuter plusieurs instructions différentes dans un seul appel :

results, cursor = conn.batch_execute(
    [
        "CREATE TABLE #BatchExec (Name NVARCHAR(50), CategoryID INT)",
        "INSERT INTO #BatchExec (Name, CategoryID) VALUES (%(name)s, %(cat)s)",
        "SELECT COUNT(*) FROM #BatchExec"
    ],
    [
        None,                                 # No params for CREATE
        {"name": "New Item", "cat": 1},       # Params for INSERT
        None                                  # No params for SELECT
    ]
)

print(f"CREATE result: {results[0]}")
print(f"INSERT affected: {results[1]} rows")
print(f"Row count: {results[2][0][0]}")

Instructions préparées

Le pilote prépare les requêtes par défaut (use_prepare=True). Lorsque vous exécutez la même chaîne SQL plusieurs fois sur le même curseur, le pilote réutilise automatiquement l’instruction préparée lors des appels suivants :

# First execution prepares the statement
cursor.execute(
    "SELECT Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = %(subcategory_id)s",
    {"subcategory_id": 1},
)
rows1 = cursor.fetchall()

# Same SQL string on same cursor → driver reuses the prepared plan
cursor.execute(
    "SELECT Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = %(subcategory_id)s",
    {"subcategory_id": 2},
)
rows2 = cursor.fetchall()

Pour éviter la préparation et utiliser l’exécution directe à la place :

cursor.execute(
    "SELECT Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = 1",
    use_prepare=False  # Uses SQLExecDirectW instead of SQLPrepareW
)

Exécution au niveau de la connexion

Pour des requêtes simples et ponctuelles, utilisez execute() directement sur la connexion :

# Creates cursor, executes, and returns cursor
cursor = conn.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
rows = cursor.fetchall()
cursor.close()

Procédures stockées

Appelez des procédures stockées en utilisant EXECUTE ou la syntaxe d’échappement ODBC {CALL} . Pour des informations sur les paramètres de sortie, les ensembles de résultats multiples et les motifs de transaction, voir Procédures stockées.

cursor.execute(
    "EXECUTE dbo.uspGetManagerEmployees @BusinessEntityID = %(business_entity_id)s",
    {"business_entity_id": 16}
)
rows = cursor.fetchall()

Définir les tailles d’entrée

Utilisez setinputsizes() pour déclarer explicitement les types de paramètres, ce qui peut améliorer les performances pour les opérations batch :

cursor.setinputsizes([
    (mssql_python.SQL_WVARCHAR, 50, 0),   # NVARCHAR(50)
    (mssql_python.SQL_INTEGER, 0, 0),     # INT
])

cursor.executemany(
    "SELECT ProductID, Name FROM Production.Product WHERE Name LIKE ? AND ProductSubcategoryID = ?",
    [("Road%", 2), ("Mountain%", 1)]
)

Gestion des erreurs

Encapsuler les opérations de base de données dans des blocs try-except :

try:
    cursor.execute("CREATE TABLE #ErrDemo (Name NVARCHAR(50) NOT NULL)")
    cursor.execute("INSERT INTO #ErrDemo (Name) VALUES (%(name)s)", {"name": None})
    conn.commit()
except mssql_python.IntegrityError as e:
    print(f"Constraint violation: {e}")
    conn.rollback()
except mssql_python.ProgrammingError as e:
    print(f"SQL error: {e}")
    conn.rollback()

Bonnes pratiques

1. Utilisez toujours des requêtes paramétrées pour éviter l’injection SQL.
2. Utilisez la copie en bloc pour les insertions en bloc au lieu de plusieurs appels à execute().
3. Commet explicitement les transactions lorsque tu désactives l’autocommit.
4. Fermez les curseurs et les connexions une fois terminé pour libérer des ressources.
5. Utilisez des gestionnaires de contexte pour le nettoyage automatique des ressources :

with mssql_python.connect(connection_string) as conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT TOP 5 Name, ListPrice FROM Production.Product")
        rows = cursor.fetchall()
# Connection and cursor automatically closed

Contenu connexe

• Récupérer des données
• Mappages de types de données
• Gestion des transactions
• Gestion du curseur