Gérer les curseurs et les ensembles de résultats

Le pilote mssql-python fournit des objets curseur pour exécuter des requêtes, gérer plusieurs ensembles de résultats et gérer efficacement la mémoire.

Notions de base du curseur

Créer et utiliser des curseurs

Appelez conn.cursor() pour créer un curseur, puis utilisez execute() et récupérez des méthodes pour lancer des requêtes et récupérer les résultats :

import mssql_python

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

# Create cursor
cursor = conn.cursor()

# Execute query
cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")

# Process results
for row in cursor:
    print(row.Name)

# Close cursor when done
cursor.close()

Modèle de gestionnaire de contexte

Utilisez cette with instruction pour implémenter un gestionnaire de contexte pour le nettoyage automatique :

with mssql_python.connect(connection_string) as conn:
    with conn.cursor() as cursor:
        cursor.execute("SELECT TOP 10 Name, ListPrice FROM Production.Product")
        products = cursor.fetchall()
    # Cursor automatically closed on exit
# Connection automatically closed on exit

Curseurs multiples

conn = mssql_python.connect(connection_string)

# Multiple cursors on same connection
cursor1 = conn.cursor()
cursor2 = conn.cursor()

# Fetch results completely from cursor1 before using cursor2
cursor1.execute("SELECT TOP 5 ProductID, Name FROM Production.Product")
products = cursor1.fetchall()

cursor2.execute("SELECT TOP 5 ProductCategoryID, Name FROM Production.ProductCategory")
categories = cursor2.fetchall()

cursor1.close()
cursor2.close()

Si vous devez effectuer des requêtes simultanément, utilisez plutôt des connexions séparées :

conn1 = mssql_python.connect(connection_string)
conn2 = mssql_python.connect(connection_string)

cursor1 = conn1.cursor()
cursor2 = conn2.cursor()

cursor1.execute("SELECT TOP 5 ProductID, Name FROM Production.Product")
cursor2.execute("SELECT TOP 5 ProductCategoryID, Name FROM Production.ProductCategory")

products = cursor1.fetchall()
categories = cursor2.fetchall()

cursor1.close()
cursor2.close()
conn1.close()
conn2.close()

Stratégies de récupération

Récupération complète vs récupération itérative

Utilisez fetchall() pour charger l’ensemble du jeu de résultats en mémoire en une seule fois, ou pour itérer sur le curseur afin de traiter les lignes une à la fois sans mise en mémoire tampon.

# Fetch all at once - loads entire result into memory
cursor.execute("SELECT * FROM Production.Product")
all_products = cursor.fetchall()
print(f"Loaded {len(all_products)} products")

# Iterative fetch - memory efficient
cursor.execute("SELECT * FROM Production.Product")
count = 0
for row in cursor:
    count += 1
print(f"Processed {count} products")

Récupérer par lots

Utilisez fetchmany() avec une taille de lot pour traiter de grands ensembles de résultats en blocs sans tout charger en mémoire.

def process_batch(rows):
    # Example: print each row. Replace with your own logic.
    for row in rows:
        print(row)

def fetch_in_batches(cursor, batch_size: int = 1000):
    """Fetch results in batches to manage memory."""
    while True:
        batch = cursor.fetchmany(batch_size)
        if not batch:
            break
        yield batch

cursor.execute("SELECT * FROM LargeTable")
for batch in fetch_in_batches(cursor, batch_size=5000):
    process_batch(batch)
    print(f"Processed batch of {len(batch)} rows")

Utilisez fetchval pour les valeurs uniques

Utilisez fetchval() pour des requêtes scalaires qui retournent une seule valeur. Il retourne la première colonne de la première ligne.

# Efficient for scalar queries
cursor.execute("SELECT COUNT(*) FROM Production.Product")
count = cursor.fetchval()  # Returns single value directly

cursor.execute("SELECT MAX(ListPrice) FROM Production.Product")
max_price = cursor.fetchval()

Ensembles de résultats multiples

Ensembles de résultats multiples de processus

Utilisez nextset() pour avancer au-delà de l’ensemble de résultats actuel vers le suivant après avoir récupéré toutes les lignes de l’ensemble précédent.

# Query returns multiple results
cursor.execute("""
    SELECT TOP 3 CustomerID, AccountNumber FROM Sales.Customer;
    SELECT TOP 3 SalesOrderID, OrderDate FROM Sales.SalesOrderHeader;
    SELECT TOP 3 ProductID, Name FROM Production.Product;
""")

# First result set
print("Customers:")
customers = cursor.fetchall()
for c in customers:
    print(f"  {c.AccountNumber}")

# Move to second result set
if cursor.nextset():
    print("Orders:")
    orders = cursor.fetchall()
    for o in orders:
        print(f"  Order #{o.SalesOrderID}")

# Move to third result set
if cursor.nextset():
    print("Products:")
    products = cursor.fetchall()
    for p in products:
        print(f"  {p.Name}")

Itère tous les ensembles de résultats

Répétez jusqu’à ce que nextset() renvoie False pour consommer tous les jeux de résultats provenant d’un seul appel à execute :

def process_all_result_sets(cursor):
    """Process all result sets from a query."""
    result_sets = []
    
    while True:
        # Fetch current result set
        rows = cursor.fetchall()
        result_sets.append(rows)
        
        # Try to move to next result set
        if not cursor.nextset():
            break
    
    return result_sets

cursor.execute("""
    SELECT TOP 3 ProductID, Name FROM Production.Product ORDER BY ProductID;
    SELECT TOP 3 SalesOrderID, TotalDue FROM Sales.SalesOrderHeader ORDER BY SalesOrderID;
""")
all_results = process_all_result_sets(cursor)
print(f"Retrieved {len(all_results)} result sets")

Vérifiez s’il existe d’autres ensembles de résultats

Vérifiez la valeur de retour de nextset() dans une boucle pour consommer tous les ensembles de résultats sans savoir à l’avance combien il y en a :

cursor.execute("""
    SELECT COUNT(*) AS ProductCount FROM Production.Product;
    SELECT COUNT(*) AS PersonCount FROM Person.Person;
""")

result_num = 1
while True:
    count = cursor.fetchval()
    print(f"Result set {result_num}: {count}")
    
    result_num += 1
    if not cursor.nextset():
        break

Description du curseur

Accéder aux métadonnées de la colonne

Après exécution d’une requête, cursor.description contient une séquence de tuples de 7 items — un par colonne — avec nom, code de type, taille d’affichage, taille interne, précision, échelle et nullabilité :

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 10")

# Get column information
for col in cursor.description:
    print(f"Column: {col[0]}, Type: {col[1]}")

# description structure: (name, type_code, display_size, internal_size, 
#                        precision, scale, null_ok)

Construire des gestionnaires dynamiques de résultats

Créez des gestionnaires de résultats qui fonctionnent avec toute requête en construisant la liste des colonnes à partir de cursor.description à l’exécution :

def query_to_dicts(cursor) -> list[dict]:
    """Convert query results to list of dictionaries."""
    columns = [col[0] for col in cursor.description]
    return [dict(zip(columns, row)) for row in cursor.fetchall()]

cursor.execute("SELECT ProductID, Name, ListPrice FROM Production.Product WHERE ProductID < 10")
products = query_to_dicts(cursor)
for p in products:
    print(p["Name"])

Gérer les requêtes sans résultat

cursor.description est None après des instructions non SELECT telles que INSERT, UPDATE, et DELETE. Vérifiez avant d’appeler les méthodes de récupération :

cursor.execute("CREATE TABLE #UpdDemo (Name NVARCHAR(50), Price DECIMAL(10,2), CategoryID INT)")
cursor.execute("INSERT INTO #UpdDemo VALUES ('Widget', 10.0, 5), ('Gadget', 20.0, 5)")
cursor.execute("UPDATE #UpdDemo SET Price = Price * 1.1 WHERE CategoryID = 5")

# description is None for non-SELECT statements
if cursor.description is None:
    print(f"Updated {cursor.rowcount} rows")
else:
    results = cursor.fetchall()

Nombre de lignes

Suivre les lignes affectées

Après INSERT, UPDATE, ou DELETE, cursor.rowcount retourne le nombre de lignes affectées par l’énoncé :

cursor.execute("CREATE TABLE #RowDemo (Name NVARCHAR(50), Stock INT)")
cursor.execute("INSERT INTO #RowDemo VALUES ('A', 0), ('B', 5), ('C', 0)")
cursor.execute("UPDATE #RowDemo SET Stock = -1 WHERE Stock = 0")
print(f"Rows affected: {cursor.rowcount}")

cursor.execute("DELETE FROM #RowDemo WHERE Stock = -1")
print(f"Deleted {cursor.rowcount} rows")

Gérer le nombre de lignes inconnu

# Some operations might not return row count
cursor.execute("EXEC dbo.uspGetEmployeeManagers @BusinessEntityID = 5")

if cursor.rowcount == -1:
    print("Row count not available")
else:
    print(f"Affected {cursor.rowcount} rows")

Ignorer des lignes

Utilisez « skip » comme alternative à la pagination

cursor.skip() fait avancer la position du curseur sans récupérer de lignes. Pour les grands ensembles de données, privilégiez la pagination OFFSET-FETCH au niveau SQL pour de meilleures performances :

def get_page_using_skip(cursor, page: int, page_size: int):
    """Get a page of results using skip."""
    cursor.execute("SELECT * FROM Production.Product ORDER BY ProductID")
    
    # Skip rows from previous pages
    cursor.skip((page - 1) * page_size)
    
    # Fetch this page
    return cursor.fetchmany(page_size)

# Get page 3
page_3 = get_page_using_skip(cursor, page=3, page_size=20)

Messages de diagnostic

Accéder à cursor.messages

L’attribut messages stocke des messages d’information générés lors de l’exécution de la déclaration SQL, comme décrit dans PEP 249. Ces messages comprennent la sortie des instructions PRINT et des messages RAISERROR dont le niveau de gravité est inférieur à 11.

L’attribut est une liste de tuples dans laquelle chaque tuple contient un code de type de message et le texte du message :

conn = mssql_python.connect(connection_string, autocommit=True)
cursor = conn.cursor()
cursor.execute("PRINT 'Hello world!'")
print(cursor.messages)

Output:

[('[01000] (0)', '[Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Hello world!')]

Le texte du message inclut les informations du préfixe du pilote car le pilote récupère les messages sous forme d’enregistrements de diagnostic via SQLGetDiagRec.

Capturer des messages à partir de procédures stockées

Lisez cursor.messages après l’exécution pour récupérer tout PRINT message de sortie ou d’information serveur issu de l’instruction précédente :

cursor.execute("EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = 5")
results = cursor.fetchall()

# Check for any informational messages
if cursor.messages:
    for msg_type, msg_text in cursor.messages:
        print(f"Server message: {msg_text}")

Gestion de la mémoire

Traiter efficacement les gros résultats

Récupérez par lots à l’aide de fetchmany() pour traiter des tables trop volumineuses pour être chargées entièrement en mémoire en une seule fois :

def process_large_table(cursor, batch_size: int = 10000):
    """Process large result set without loading all into memory."""
    cursor.execute("SELECT * FROM VeryLargeTable")
    
    total_processed = 0
    while True:
        rows = cursor.fetchmany(batch_size)
        if not rows:
            break
        
        for row in rows:
            process_row(row)
        
        total_processed += len(rows)
        print(f"Progress: {total_processed} rows processed")
    
    return total_processed

Traitement basé sur un générateur

Encapsulez la récupération par lots dans un générateur pour traiter une ligne à la fois tout en gardant l’utilisation de la mémoire constante, quelle que soit la taille de l’ensemble de résultats :

def row_generator(cursor, batch_size: int = 1000):
    """Generate rows from cursor without loading all."""
    while True:
        rows = cursor.fetchmany(batch_size)
        if not rows:
            break
        for row in rows:
            yield row

cursor.execute("SELECT * FROM LargeTable")
for row in row_generator(cursor, batch_size=5000):
    # Process one row at a time
    print(row)  # Replace with your own row-handling logic

Fermez rapidement les curseurs

Fermez toujours les curseurs dans un finally bloc pour libérer les ressources côté serveur même en cas d’exception :

def get_product(conn, product_id: int):
    """Get product and properly close cursor."""
    cursor = conn.cursor()
    try:
        cursor.execute(
            "SELECT * FROM Production.Product WHERE ProductID = %(id)s",
            {"id": product_id}
        )
        return cursor.fetchone()
    finally:
        cursor.close()

Gestion de l’état du curseur

Vérifiez si le curseur contient des données

Vérifiez si une requête renvoie des lignes en vérifiant si fetchone() renvoie None:

cursor.execute("SELECT ProductID, Name FROM Production.Product WHERE ProductID = 999")
row = cursor.fetchone()

if row is None:
    print("Product not found")
else:
    print(f"Found: {row.Name}")

Réutiliser les curseurs

Un seul curseur peut exécuter plusieurs requêtes séquentiellement. Chaque execute() appel remplace l’ensemble de résultats précédent :

cursor = conn.cursor()

# Execute multiple queries with same cursor
cursor.execute("SELECT TOP 5 * FROM Sales.Customer")
customers = cursor.fetchall()

cursor.execute("SELECT TOP 5 * FROM Production.Product")
products = cursor.fetchall()

cursor.execute("SELECT TOP 5 * FROM Sales.SalesOrderHeader")
orders = cursor.fetchall()

cursor.close()

Bonnes pratiques

Modèle : classe utilitaire de curseur

Encapsulez la gestion du cycle de vie du curseur dans une classe helper pour réduire le code standard à travers votre application :

class CursorManager:
    """Helper for managing cursor lifecycle."""
    
    def __init__(self, connection):
        self.conn = connection
    
    def execute_and_fetch(self, query: str, params: dict = None) -> list:
        """Execute query and return all results."""
        cursor = self.conn.cursor()
        try:
            cursor.execute(query, params or {})
            return cursor.fetchall()
        finally:
            cursor.close()
    
    def execute_scalar(self, query: str, params: dict = None):
        """Execute query and return single value."""
        cursor = self.conn.cursor()
        try:
            cursor.execute(query, params or {})
            return cursor.fetchval()
        finally:
            cursor.close()
    
    def execute_non_query(self, query: str, params: dict = None) -> int:
        """Execute non-SELECT and return row count."""
        cursor = self.conn.cursor()
        try:
            cursor.execute(query, params or {})
            return cursor.rowcount
        finally:
            cursor.close()

# Usage
db = CursorManager(conn)
products = db.execute_and_fetch("SELECT TOP 5 Name FROM Production.Product")
count = db.execute_scalar("SELECT COUNT(*) FROM Production.Product")

db.execute_non_query("CREATE TABLE #Logs (LogID INT, Age INT)")
db.execute_non_query("INSERT INTO #Logs VALUES (1, 45), (2, 20), (3, 60)")
affected = db.execute_non_query("DELETE FROM #Logs WHERE Age > 30")

Ne laissez pas les curseurs ouverts

Un curseur qui n’est pas explicitement fermé retient les ressources côté serveur jusqu’à la fermeture de la connexion. Utilisez try/finally pour garantir le nettoyage :

# Bad: cursor left open
def get_data_bad(conn):
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM Data")
    return cursor.fetchall()
    # Cursor never closed!

# Good: always close cursor
def get_data_good(conn):
    cursor = conn.cursor()
    try:
        cursor.execute("SELECT * FROM Data")
        return cursor.fetchall()
    finally:
        cursor.close()

Synchronisez la durée de vie du curseur à l’opération

Créez des curseurs éphémères pour des opérations uniques. Réutilisez le même curseur uniquement pour une séquence d’opérations liées :

# Short-lived cursor for simple query
def get_user_count(conn) -> int:
    cursor = conn.cursor()
    try:
        cursor.execute("SELECT COUNT(*) FROM Person.Person")
        return cursor.fetchval()
    finally:
        cursor.close()

# Reuse cursor for related operations
def update_inventory(conn, items: list):
    cursor = conn.cursor()
    try:
        for item in items:
            cursor.execute(
                "UPDATE Inventory SET Quantity = %(qty)s WHERE ProductID = %(id)s",
                item
            )
        conn.commit()
    finally:
        cursor.close()

Contenu connexe

• Exécution des requêtes
• Récupération des données
• Regroupement de connexions
• Pagination