Créer des requêtes paramétrées

Les requêtes paramétrées sont essentielles pour :

Sécurité : prévenir les attaques par injection SQL
Performance : Activer la réutilisation du plan de requête
Correction : Gestion appropriée des caractères et types de données spéciaux

Le pilote mssql-python utilise par défaut le pyformat style de paramètres avec %(name)s des placeholders, mais prend aussi en charge d’autres styles de paramètres si vous préférez ce format.

Requêtes paramétrées de base

Paramètres nommés

Utilisez des placeholders nommés pour transmettre des paramètres aux requêtes :

import mssql_python

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

# Single parameter
cursor.execute(
    "SELECT * FROM Production.Product WHERE ProductSubcategoryID = %(category)s",
    {"category": 5}
)

# Multiple parameters
cursor.execute(
    "SELECT * FROM Production.Product WHERE ProductSubcategoryID = %(cat)s AND ListPrice > %(price)s",
    {"cat": 5, "price": 10.00}
)

Réutilisation des paramètres

Vous pouvez consulter le même paramètre plusieurs fois :

cursor.execute("""
    SELECT * FROM Production.Product 
    WHERE (Name LIKE %(search)s OR ProductNumber LIKE %(search)s)
    AND ProductSubcategoryID = %(cat)s
""", {"search": "%Road%", "cat": 2})

Types de données dans des paramètres

Paramètres de chaîne

Les chaînes sont automatiquement mises entre guillemets et les caractères spéciaux sont échappés de manière sécurisée :

# Strings are automatically quoted
cursor.execute(
    "SELECT * FROM Person.EmailAddress WHERE EmailAddress = %(email)s",
    {"email": "ken0@adventure-works.com"}
)

# Special characters are escaped
cursor.execute(
    "SELECT * FROM Person.Person WHERE LastName = %(name)s",
    {"name": "O'Brien"}  # Apostrophe handled safely
)

Paramètres numériques

Passez les valeurs numériques sous forme d’entiers, décimaux ou flottants selon la précision requise :

from decimal import Decimal

# Integer
cursor.execute("SELECT * FROM Production.Product WHERE ProductID = %(id)s", {"id": 42})

# Decimal for financial precision
cursor.execute(
    "SELECT * FROM Production.Product WHERE ListPrice >= %(min)s AND ListPrice <= %(max)s",
    {"min": Decimal("10.00"), "max": Decimal("100.00")}
)

# Float
cursor.execute(
    "SELECT * FROM Production.Product WHERE Weight > %(threshold)s",
    {"threshold": 15.0}
)

Paramètres date/heure

Utilisez le module de datetime Python pour passer les valeurs de date, de date-heure et d'heure :

from datetime import date, datetime, time

# Date
cursor.execute(
    "SELECT * FROM Sales.SalesOrderHeader WHERE OrderDate >= %(date)s AND OrderDate < DATEADD(day, 1, %(date)s)",
    {"date": date(2014, 3, 15)}
)

# Datetime
cursor.execute(
    "SELECT * FROM Sales.SalesOrderHeader WHERE ModifiedDate >= %(start)s AND ModifiedDate < %(end)s",
    {"start": datetime(2014, 3, 1), "end": datetime(2014, 4, 1)}
)

# Time
cursor.execute(
    "SELECT * FROM HumanResources.Shift WHERE StartTime >= %(time)s",
    {"time": time(9, 0, 0)}
)

Aucun pour NULL

Passer None pour insérer ou mettre à jour les valeurs NULL dans la base de données :

# Insert NULL
cursor.execute("""
    CREATE TABLE #NullDemo (ID INT IDENTITY, Name NVARCHAR(50), Email NVARCHAR(100))
""")
cursor.execute(
    "INSERT INTO #NullDemo (Name, Email) VALUES (%(name)s, %(email)s)",
    {"name": "Guest", "email": None}
)

# Query with NULL
cursor.execute(
    "UPDATE #NullDemo SET Email = %(email)s WHERE ID = %(id)s",
    {"email": None, "id": 1}
)

Paramètres binaires

Insérer des données binaires sous forme d’objets octets :

# Binary data
hash_value = b'\x00\x01\x02\x03'
cursor.execute("""
    CREATE TABLE #HashDemo (ID INT IDENTITY, DocumentHash VARBINARY(256))
""")
cursor.execute(
    "INSERT INTO #HashDemo (DocumentHash) VALUES (%(hash)s)",
    {"hash": hash_value}
)

Construire des requêtes dynamiques

Clauses WHERE conditionnelles

Construisez dynamiquement les clauses WHERE en fonction de critères de recherche optionnels :

def search_products(cursor, name: str | None = None, 
                   category: int | None = None,
                   min_price: float | None = None) -> list:
    """Build query with optional conditions."""
    conditions = []
    params = {}
    
    if name:
        conditions.append("Name LIKE %(name)s")
        params["name"] = f"%{name}%"
    
    if category:
        conditions.append("ProductSubcategoryID = %(category)s")
        params["category"] = category
    
    if min_price is not None:
        conditions.append("ListPrice >= %(min_price)s")
        params["min_price"] = min_price
    
    query = "SELECT TOP 10 * FROM Production.Product"
    if conditions:
        query += " WHERE " + " AND ".join(conditions)
    
    cursor.execute(query, params)
    return cursor.fetchall()

# Usage
products = search_products(cursor, name="Road", min_price=10.0)

Clause IN avec plusieurs valeurs

Construis la IN clause dynamiquement avec un marque-place pour chaque valeur. N’utilisez jamais la mise en forme de chaîne pour injecter des valeurs directement :

def get_products_by_ids(cursor, product_ids: list[int]) -> list:
    """Query with IN clause using qmark (?) placeholders."""
    if not product_ids:
        return []

    placeholders = ", ".join("?" for _ in product_ids)
    query = f"SELECT * FROM Production.Product WHERE ProductID IN ({placeholders})"
    cursor.execute(query, tuple(product_ids))
    return cursor.fetchall()

# Usage
products = get_products_by_ids(cursor, [1, 5, 10, 15])

Le même motif fonctionne avec les placeholders pyformat (%(name)s) :

def get_products_by_ids(cursor, product_ids: list[int]) -> list:
    """Query with IN clause using pyformat placeholders."""
    if not product_ids:
        return []
    
    # Create named parameters for each ID
    params = {f"id{i}": id for i, id in enumerate(product_ids)}
    placeholders = ", ".join(f"%(id{i})s" for i in range(len(product_ids)))
    
    query = f"SELECT * FROM Production.Product WHERE ProductID IN ({placeholders})"
    cursor.execute(query, params)
    return cursor.fetchall()

# Usage
products = get_products_by_ids(cursor, [1, 5, 10, 15])

Sélection dynamique de colonnes

Utilisez une liste de permis pour valider les colonnes avant de construire dynamiquement des listes SELECT, tout en conservant les valeurs des filtres comme paramètres :

def get_employee(cursor, employee_id: int, columns: list[str] | None = None) -> dict:
    """Get employee with specified columns."""
    # Allow list of permitted columns
    allowed = {"BusinessEntityID", "LoginID", "JobTitle", "HireDate", "SalariedFlag"}
    
    if columns:
        # Validate columns against allow list
        safe_columns = [c for c in columns if c in allowed]
        if not safe_columns:
            raise ValueError("No valid columns specified")
        column_list = ", ".join(safe_columns)
    else:
        column_list = "*"
    
    # ID is always a parameter, never interpolated
    query = f"SELECT {column_list} FROM HumanResources.Employee WHERE BusinessEntityID = %(id)s"
    cursor.execute(query, {"id": employee_id})
    return cursor.fetchone()

Ordre de tri

Utilisez une liste de permis pour valider les colonnes de tri avant l’interpolation :

def get_products_sorted(cursor, sort_by: str = "Name", 
                       descending: bool = False) -> list:
    """Get products with validated sort order."""
    # Allow list of permitted sort columns
    allowed_sorts = {"Name", "ListPrice", "SellStartDate", "ProductID"}
    
    if sort_by not in allowed_sorts:
        sort_by = "Name"  # Default
    
    direction = "DESC" if descending else "ASC"
    
    # sort_by and direction are validated, safe to interpolate
    query = f"SELECT TOP 10 * FROM Production.Product ORDER BY {sort_by} {direction}"
    cursor.execute(query)
    return cursor.fetchall()

Opérations INSERT

Insertion unique

Insérer une seule ligne avec des valeurs paramétrées :

cursor.execute("""
    CREATE TABLE #ParamInsert (ID INT IDENTITY, Name NVARCHAR(50), Price DECIMAL(10,2), CategoryID INT)
""")
cursor.execute("""
    INSERT INTO #ParamInsert (Name, Price, CategoryID)
    VALUES (%(name)s, %(price)s, %(category)s)
""", {"name": "New Widget", "price": 29.99, "category": 5})
conn.commit()

Insertion avec retour de l’identifiant

Utilisez OUTPUT pour récupérer la valeur d’identité générée après avoir inséré une nouvelle ligne :

cursor.execute("""
    CREATE TABLE #IdentDemo (ProductID INT IDENTITY, Name NVARCHAR(50), Price DECIMAL(10,2), CategoryID INT)
""")
cursor.execute("""
    INSERT INTO #IdentDemo (Name, Price, CategoryID)
    OUTPUT INSERTED.ProductID
    VALUES (%(name)s, %(price)s, %(category)s)
""", {"name": "New Widget", "price": 29.99, "category": 5})

new_id = cursor.fetchval()
conn.commit()
print(f"Created product with ID: {new_id}")

Insertion par lot avec executemany

À utiliser executemany() pour insérer plusieurs lignes efficacement avec une seule instruction paramétrée :

Tip

Pour les gros volumes, bulkcopy() est plus rapide que executemany() car il utilise le protocole d’insertion en bloc au lieu d’instructions INSERT individuelles. Voir Copie en bloc.

cursor.execute("""
    CREATE TABLE #BatchDemo (ID INT IDENTITY, Name NVARCHAR(50), Price DECIMAL(10,2), CategoryID INT)
""")

products = [
    {"name": "Widget A", "price": 19.99, "cat": 1},
    {"name": "Widget B", "price": 29.99, "cat": 1},
    {"name": "Widget C", "price": 39.99, "cat": 2},
]

cursor.executemany("""
    INSERT INTO #BatchDemo (Name, Price, CategoryID)
    VALUES (%(name)s, %(price)s, %(cat)s)
""", products)
conn.commit()

Opérations UPDATE

Mettez à jour des lignes simples ou multiples en fonction des conditions à l’aide de clauses WHERE paramétrées :

# Create temp table with sample data
cursor.execute("""
    CREATE TABLE #UpdDemo (
        ID INT IDENTITY, Name NVARCHAR(50),
        Price DECIMAL(10,2), CategoryID INT, ModifiedAt DATETIME
    )
""")
cursor.execute("""
    INSERT INTO #UpdDemo (Name, Price, CategoryID)
    VALUES ('Widget X', 25.00, 5), ('Widget Y', 30.00, 5), ('Gadget Z', 50.00, 3)
""")

# Single row update
cursor.execute("""
    UPDATE #UpdDemo 
    SET Price = %(price)s, ModifiedAt = %(modified)s
    WHERE ID = %(id)s
""", {"price": 34.99, "modified": datetime.now(), "id": 1})

# Conditional update
cursor.execute("""
    UPDATE #UpdDemo 
    SET Price = Price * %(multiplier)s
    WHERE CategoryID = %(category)s
""", {"multiplier": 1.1, "category": 5})
conn.commit()

Opérations DELETE

Supprimez les lignes des tables en fonction des conditions de filtre paramétrées :

# Create temp table with sample data
cursor.execute("""
    CREATE TABLE #DelDemo (
        ID INT IDENTITY, Name NVARCHAR(50), Status NVARCHAR(20), OrderDate DATE
    )
""")
cursor.execute("""
    INSERT INTO #DelDemo (Name, Status, OrderDate)
    VALUES ('Order1', 'Active', '2024-06-01'), ('Order2', 'Cancelled', '2022-05-01'),
           ('Order3', 'Cancelled', '2022-11-01')
""")

# Delete single row
cursor.execute(
    "DELETE FROM #DelDemo WHERE ID = %(id)s",
    {"id": 1}
)

# Delete with conditions
cursor.execute("""
    DELETE FROM #DelDemo 
    WHERE Status = %(status)s AND OrderDate < %(date)s
""", {"status": "Cancelled", "date": date(2023, 1, 1)})

conn.commit()

Considérations relatives à la sécurité

Ne jamais interpoler les entrées de l’utilisateur

Utilisez toujours des paramètres pour échapper à l’entrée de l’utilisateur en toute sécurité :

# DANGEROUS - SQL injection vulnerability!
user_input = "'; DROP TABLE Users;--"
query = f"SELECT * FROM Person.Person WHERE LastName = '{user_input}'"  # DON'T DO THIS

# SAFE - always use parameters
cursor.execute(
    "SELECT * FROM Person.Person WHERE LastName = %(name)s",
    {"name": user_input}  # Input is safely escaped
)

Valider les noms des tables et des colonnes

Utilisez les listes de permis pour valider les identifiants de tables et de colonnes qui ne peuvent pas être paramétrés :

def query_table(cursor, table: str, columns: list[str]):
    """Query with validated table and column names."""
    # Allow list of permitted tables
    allowed_tables = {"Person.Person", "Production.Product", "Sales.SalesOrderHeader"}
    if table not in allowed_tables:
        raise ValueError(f"Invalid table: {table}")
    
    # Allow list of permitted columns per table
    allowed_columns = {
        "Person.Person": {"BusinessEntityID", "FirstName", "LastName"},
        "Production.Product": {"ProductID", "Name", "ListPrice"},
        "Sales.SalesOrderHeader": {"SalesOrderID", "CustomerID", "TotalDue"},
    }
    
    safe_columns = [c for c in columns if c in allowed_columns.get(table, set())]
    if not safe_columns:
        raise ValueError("No valid columns")
    
    # Safe to interpolate after validation
    query = f"SELECT TOP 5 {', '.join(safe_columns)} FROM {table}"
    cursor.execute(query)
    return cursor.fetchall()

Utiliser des procédures stockées pour des opérations complexes

Les procédures stockées ajoutent une couche supplémentaire de protection et permettent l’exécution d’une logique métier complexe côté serveur :

cursor.execute("""
    EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = %(id)s
""", {"id": 5})
rows = cursor.fetchall()

Avantages en matière de performances

Mise en cache des plans de requête

Lorsque vous utilisez des requêtes paramétrées, SQL Server réutilise le même plan d’exécution sur différentes valeurs de paramètres au lieu de compiler un nouveau plan pour chaque requête.

for product_id in range(1, 100):
    cursor.execute(
        "SELECT * FROM Production.Product WHERE ProductID = %(id)s",
        {"id": product_id}
    )

Instructions préparées

Utilisez des instructions préparées pour les requêtes que vous lancez fréquemment. Le pilote prépare automatiquement les instructions, donc exécuter le même modèle de requête avec des paramètres différents bénéficie de la préparation.

query = "SELECT Name, ListPrice FROM Production.Product WHERE ProductSubcategoryID = %(cat)s"

for category in [1, 2, 3, 4, 5]:
    cursor.execute(query, {"cat": category})
    products = cursor.fetchall()

Commentaires

Cette page a-t-elle été utile ?

Last updated on 2026-07-22