Cursor und Ergebnismengen verwalten

Der mssql-python-Treiber stellt Cursorobjekte zur Verfügung, um Abfragen auszuführen, mehrere Ergebnissets zu verwalten und den Speicher effizient zu verwalten.

Cursor-Grundlagen

Erstelle und benutze Cursor

Rufen Sie auf, conn.cursor() um einen Cursor zu erstellen, dann verwenden execute() und holen Sie Methoden, um Abfragen auszuführen und Ergebnisse abzurufen:

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

Kontextmanager-Muster

Verwenden Sie die Anweisung with , um einen Kontextmanager für die automatische Bereinigung zu implementieren:

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

Mehrere Cursor

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

Wenn Sie Abfragen gleichzeitig ausführen müssen, verwenden Sie stattdessen separate Verbindungen:

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

Fetch-Strategien

Alles abrufen vs. iteratives Abrufen

Verwenden Sie fetchall(), um die gesamte Ergebnismenge vollständig in den Speicher zu laden, oder durchlaufen Sie den Cursor, um Zeilen einzeln ohne Pufferung zu verarbeiten.

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

Stapelweise abrufen

Verwenden Sie fetchmany() mit einer Batch-Größe, um große Ergebnismengen in Blöcken zu verarbeiten, ohne alles vollständig in den Speicher zu laden.

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

Verwenden Sie fetchval für einzelne Werte

Verwenden Sie fetchval() für skalare Abfragen, die einen einzelnen Wert zurückgeben. Es gibt die erste Spalte der ersten Zeile zurück.

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

Mehrere Ergebnismengen

Mehrere Ergebnismengen verarbeiten

Verwenden Sie nextset(), um nach dem Abrufen aller Zeilen aus der vorherigen Ergebnismenge zur nächsten Ergebnismenge weiterzugehen.

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

Iteriere alle Ergebnismengen

Wiederholen Sie den Vorgang, bis nextset()False zurückgibt, um alle Ergebnismengen eines einzelnen Ausführungsaufrufs zu verarbeiten:

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

Überprüfen Sie, ob es weitere Ergebnismengen gibt

Überprüfen Sie den Rückgabewert von nextset() in einer Schleife, um alle Ergebnismengen zu konsumieren, ohne im Voraus zu wissen, wie viele es sind:

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

Cursorbeschreibung

Zugreifen auf Spalten-Metadaten

Nach der Ausführung einer Abfrage cursor.description enthält eine Folge von 7-Item-Tupeln – eines pro Spalte – mit Name, Typcode, Anzeigegröße, interner Größe, Genauigkeit, Skalierung und Nullierbarkeit:

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)

Dynamische Ergebnis-Handler bauen

Erstelle Ergebnishandler, die mit jeder Abfrage arbeiten, indem sie zur Laufzeit die Spaltenliste aus cursor.description erstellen:

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

Bearbeitung von Abfragen ohne Ergebnisse

cursor.description ist None nach Nicht-SELECT-Anweisungen wie INSERT, UPDATE, und DELETE. Überprüfen Sie es, bevor Sie die Fetch-Methoden aufrufen:

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

Zeilenanzahl

Betroffene Zeilen nachverfolgen

Nach INSERT, UPDATE, oder DELETE, cursor.rowcount gibt die Anzahl der von der Aussage betroffenen Zeilen zurück:

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

Umgang mit unbekannter Zeilenanzahl

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

Zeilen überspringen

Verwenden Sie Überspringen als Alternative zur Paginierung

cursor.skip() rückt die Cursor-Position vor, ohne Zeilen abzurufen. Verwenden Sie bei großen Datensätzen vorzugsweise die OFFSET-FETCH-Paginierung auf SQL-Ebene, um eine bessere Leistung zu erzielen:

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)

Diagnostische Nachrichten

Zugriff auf cursor.messages

Das Attribut messages speichert Informationsnachrichten, die während der Ausführung von SQL-Anweisungen generiert werden, wie in PEP 249 beschrieben. Diese Nachrichten umfassen Ausgaben aus PRINT-Anweisungen und RAISERROR mit Schweregraden unter 11.

Das Attribut ist eine Liste von Tupeln, wobei jedes Tupel einen Nachrichtentypcode und den Nachrichtentext enthält:

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

Der Nachrichtentext enthält Informationen zum Treiberpräfix, da der Treiber Nachrichten als Diagnosedatensätze über SQLGetDiagRecabruft.

Erfassung von Nachrichten aus gespeicherten Prozeduren

Lesen Sie cursor.messages nach der Ausführung, um Ausgaben oder informative Servermeldungen der vorherigen Anweisung aus PRINT abzurufen:

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

Speicherverwaltung

Verarbeiten Sie große Ergebnisse effizient

Rufen Sie mithilfe von fetchmany() stapelweise Daten ab, um Tabellen zu verarbeiten, die zu groß sind, um auf einmal in den Speicher geladen zu werden:

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

Generatorbasierte Verarbeitung

Kapseln Sie das Batch-Fetching in einen Generator, um jeweils eine Zeile zu verarbeiten, während der Speicherverbrauch unabhängig von der Größe der Ergebnismenge konstant bleibt:

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

Cursor sofort schließen

Schließen Sie immer die Cursor in einem finally Block, um serverseitige Ressourcen freizugeben, selbst wenn eine Ausnahme auftritt:

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

Cursor-Zustandsverwaltung

Überprüfe, ob der Cursor Daten enthält

Testen Sie, ob eine Abfrage Zeilen zurückgibt, indem Sie prüfen, ob fetchone()None zurückgibt:

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

Cursor wiederverwenden

Ein einzelner Cursor kann mehrere Abfragen nacheinander ausführen. Jeder execute() Aufruf ersetzt die vorherige Ergebnismenge:

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

Bewährte Methoden

Muster: Cursor-Helfer-Klasse

Kapseln Sie das Cursor-Lebenszyklusmanagement in einer Hilfsklasse, um Boilerplate in Ihrer Anwendung zu reduzieren:

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

Lass die Cursor nicht offen

Ein Cursor, der nicht explizit geschlossen ist, hält serverseitige Ressourcen, bis die Verbindung geschlossen wird. Verwenden Sie try/finally, um die Bereinigung sicherzustellen:

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

Cursor-Lebensdauer an Betrieb anpassen

Erstelle kurzlebige Cursor für einzelne Operationen. Verwenden Sie denselben Cursor nur für eine Folge verwandter Operationen:

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

Verwandte Inhalte

• Abfragen ausführen
• Datenabruf
• Verbindungspooling
• Seitennummerierung