Crea query parametrizzate

Le query parametrizzate sono essenziali per:

Sicurezza: prevenire attacchi di SQL injection
Prestazioni: Abilitare il riutilizzo del piano di query
Correttezza: Gestione corretta di caratteri e tipi di dati speciali

Il driver mssql-python usa lo pyformat stile dei parametri con %(name)s segnaposto di default, ma supporta anche altri stili di parametri se preferisci quel formato.

Query parametrizzate di base

Parametri denominati

Usa segnaposto nominati per passare i parametri alle query:

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

Riutilizzo dei parametri

Puoi fare riferimento allo stesso parametro più volte:

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

Tipi di dati nei parametri

Parametri stringa

Le stringhe vengono automaticamente citate e i caratteri speciali vengono liberati in sicurezza:

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

Parametri numerici

Passa i valori numerici come interi, decimali o float a seconda della precisione richiesta:

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

Parametri data/ora

Usa il modulo di datetime Python per passare i valori di data, data-ora e ora:

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

Nessuno per NULL

Pass None per inserire o aggiornare i valori NULL nel 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}
)

Parametri binari

Inserisci dati binari come oggetti bytes:

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

Crea query dinamiche

Proposizioni WHERE condizionate

Costruisci le clausole WHERE dinamicamente basandosi su criteri di ricerca opzionali:

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)

clausola IN con valori multipli

Costruisci la IN clausola dinamicamente con un segnaposto per ogni valore. Non usare mai la formattazione delle stringhe per iniettare valori direttamente:

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

Lo stesso schema funziona con i segnaposto 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])

Selezione dinamica delle colonne

Usa una lista di permessi per validare le colonne prima di costruire dinamicamente liste SELECT, mantenendo i valori dei filtri come parametri:

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

Ordine di ordinamento

Usa una lista di permessi per validare le colonne di ordinamento prima dell'interpolazione:

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 Operazioni

Inserto singolo

Inserisci una singola riga con valori parametrizzati:

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

Inserimento con restituzione dell'ID

Usa OUTPUT per recuperare il valore identità generato dopo aver inserito una nuova riga:

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

Inserimento in batch con executemany

Usa executemany() per inserire più righe in modo efficiente con una singola istruzione parametrizzata:

Tip

Per grandi volumi, bulkcopy() è più veloce di executemany() perché utilizza il protocollo bulk insert invece di singole istruzioni INSERT. Vedi copia in blocco.

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 Operazioni

Aggiornare righe singole o più righe in base alle condizioni usando clausole WHERE parametrizzate:

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

Elimina le righe dalle tabelle in base alle condizioni del filtro parametrizzate:

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

Considerazioni relative alla sicurezza

Mai interpolare l'input dell'utente

Usa sempre parametri per sfuggire in sicurezza agli input dell'utente:

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

Valida i nomi di tabelle e colonne

Usa liste di permessi per convalidare identificatori di tabelle e colonne che non possono essere parametrizzati:

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

Utilizzare le stored procedure per le operazioni complesse

Le stored procedure aggiungono un ulteriore livello di protezione e permettono di eseguire una logica aziendale complessa lato server:

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

Vantaggi delle prestazioni

Memorizzazione nella cache dei piani di query

Quando usi query parametrizzate, SQL Server riutilizza lo stesso piano di esecuzione su valori di parametri diversi invece di compilare un nuovo piano per ogni query.

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

Istruzioni preparate

Usa istruzioni preparate per le query che esegui spesso. Il driver prepara automaticamente le sentenze, quindi eseguire lo stesso template di query con parametri diversi beneficia della preparazione.

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

Commenti e suggerimenti

Questa pagina è stata utile?

Last updated on 2026-07-22