Bouw geparametriseerde queries

Geparametriseerde queries zijn essentieel voor:

• Beveiliging: Het voorkomen van SQL-injectieaanvallen
• Prestaties: Hergebruik van queryplan's mogelijk maken
• Correctheid: Correcte behandeling van speciale tekens en datatypen

De mssql-python driver gebruikt standaard de pyformat parameterstijl met %(name)s placeholders, maar ondersteunt ook andere parameterstijlen als je dat formaat prefereert.

Basis geparametriseerde queries

Parameters met naam

Gebruik benoemde placeholders om parameters door te geven aan queries:

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

Hergebruik van parameters

Je kunt dezelfde parameter meerdere keren raadplegen:

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

Datatypen in parameters

Stringparameters

Snaren worden automatisch aangehaald en speciale tekens worden veilig ontsnapt:

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

Numerieke parameters

Geef numerieke waarden door als gehele getallen, decimalen of floats, afhankelijk van de vereiste precisie:

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

Datum/tijdparameters

Gebruik de datetime module van Python om datum-, datum- en tijdwaarden door te geven:

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

Geen voor NULL

Geef door None om NULL-waarden in te voegen of te updaten in de database:

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

Binaire parameters

Voeg binaire gegevens in als bytesobjecten:

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

Maak dynamische query's

Conditionele WHERE-clausules

Bouw WHERE-clausules dynamisch op basis van optionele zoekcriteria:

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)

IN-clausule met meerdere waarden

Stel de IN-clausule dynamisch samen met een plaatsaanduiding voor elke waarde. Gebruik nooit stringopmaak om waarden direct te injecteren:

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

Hetzelfde patroon werkt met pyformat (%(name)s) plaatshouders:

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

Dynamische kolomselectie

Gebruik een toestaanlijst om kolommen te valideren voordat je dynamisch SELECT-lijsten bouwt, terwijl filterwaarden als parameters worden bewaard:

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

Sorteervolgorde

Gebruik een toestemmingslijst om sorteerkolommen te valideren vóór interpolatie:

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

INSERT bewerkingen

Enkel inzetstuk

Voeg een enkele rij in met geparametriseerde waarden:

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

Insert met identiteitsretour

Gebruik OUTPUT om de gegenereerde identiteitswaarde op te halen na het invoegen van een nieuwe rij:

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

Bulksgewijs invoegen met executemany

Gebruik executemany() om efficiënt meerdere rijen in te voegen met één geparametriseerde uitspraak:

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

UPDATE bewerkingen

Werk enkele rijen of meerdere rijen bij op basis van voorwaarden met behulp van geparametriseerde WHERE-clausules:

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

DELETE bewerkingen

Verwijder rijen uit tabellen op basis van geparametriseerde filtercondities:

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

Beveiligingsoverwegingen

Interpoleer nooit gebruikersinvoer

Gebruik altijd parameters om gebruikersinvoer veilig te ontwijken:

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

Valideer tabel- en kolomnamen

Gebruik toestemmingslijsten om tabel- en kolomidentificaties te valideren die niet geparametriseerd kunnen worden:

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

Gebruik opgeslagen procedures voor complexe bewerkingen

Opgeslagen procedures voegen een extra beschermingslaag toe en maken het mogelijk om complexe bedrijfslogica serverzijde uit te voeren:

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

Prestatievoordelen

Queryplan-caching

Wanneer je geparametriseerde queries gebruikt, hergebruikt SQL Server hetzelfde uitvoeringsplan over verschillende parameterwaarden in plaats van voor elke query een nieuw plan te compileren.

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

Voorbereide statements

Gebruik voorbereide statements voor zoekopdrachten die je vaak uitvoert. De driver bereidt automatisch statements voor, dus het uitvoeren van hetzelfde querytemplate met verschillende parameters profiteert van de voorbereiding.

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

Verwante onderwerpen

• Uitvoeren van queries
• Foutafhandeling
• Stored procedures