A kurzorok és eredményhalmazok kezelése

Az mssql-python illesztőprogram kurzorobjektumokat biztosít lekérdezések végrehajtásához, több eredményhalmazt kezelésére és a memória hatékony kezelésére.

Kurzor alapjai

Hozzon létre és használjon kurzorokat

A kurzor létrehozásához hívd meg a conn.cursor() elemet, majd a lekérdezések futtatásához és az eredmények lekéréséhez használd a execute() elemet és a fetch metódusokat:

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

Kontextuskezelő minta

Használd az with utasítást egy kontextusmenedzser megvalósításához automatikus tisztításhoz:

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

Több kurzor

Important

Az mssql-python illesztőprogram nem támogatja a Több Aktív Eredményhalmazt (MARS). Egyetlen kapcsolaton több kurzort is létrehozhatsz, de egyszerre csak egy kurzor lehet aktív lekérdezést létrehozni. Mindig lekérd az összes eredményt egy kurzorból, mielőtt ugyanazon a kapcsolaton egy másik kurzoron futtatnád.

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

Ha egyszerre kell lekérdezéseket futtatni, használj külön kapcsolatokat:

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

Lekérési stratégiák

Összes lekérése szemben az iteratív lekéréssel

Használd fetchall() az egész eredményhalmazt egyszerre betöltésére a memóriába, vagy iteráld a kurzort, hogy sorokat egyenként feldolgozzd pufferlenés nélkül.

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

Csomagokban hozd be

Használd a(z) fetchmany() elemet kötegmérettel a nagy eredményhalmazok darabokban történő feldolgozásához anélkül, hogy mindent a memóriába kellene tölteni.

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

Használd fetchval egyes értékekhez

Skaláris lekérdezésekhez használják fetchval() , amelyek egyetlen értéket adnak vissza. Az első sor első oszlopát adja vissza.

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

Több eredményhalmaz

Több eredményhalmaz feldolgozása

A nextset() használatával az előző eredményhalmaz összes sorának lekérdezése után a következő eredményhalmazra léphetsz.

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

Ismételd az összes eredményhalmazt

Ismételje addig, amíg a(z) nextset() a(z) False értéket nem adja vissza, hogy egyetlen execute hívásból az összes eredményhalmazt felhasználja:

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

Ellenőrizd, van-e további eredményhalmaz.

Ellenőrizd a(z) nextset() visszatérési értékét egy ciklusban, hogy előzetes tudás nélkül az összes eredményhalmazt feldolgozhasd.

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

Kurzor leírás

Hozzáférés oszlop metaadataihoz

A lekérdezés végrehajtása után a cursor.description 7 elemű rekordok sorozatát tartalmazza — oszloponként egyet — a következőkkel: név, típuskód, megjelenítési méret, belső méret, pontosság, skála és nullabilitás:

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)

Dinamikus eredménykezelők építése

Hozzon létre olyan eredménykezelőket, amelyek bármilyen lekérdezéssel működnek, úgy, hogy az oszloplistát futásidőben a(z) cursor.description alapján állítják össze:

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

Kezeld az eredménytelen lekérdezéseket

cursor.description None nem SELECT utasítások, például INSERT, UPDATE és DELETE után. Ellenőrizd, mielőtt hívnánk a fetch metóduszokat:

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

Sorok száma

Érintett sorok nyomon követése

Miután INSERT, UPDATE, vagy DELETE, cursor.rowcount visszaadja az állítás által érintett sorok számát:

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

Ismeretlen számú sor kezelése

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

Sorok kihagyása

Használd a skip-et oldalozási alternatívához

cursor.skip() előrébb lépteti a kurzort sorok lekérése nélkül. Nagy adathalmazokhoz az SQL szintű OFFSET-FETCH oldalozást preferáljuk a jobb teljesítmény érdekében:

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)

Megjegyzés:

Nagy adathalmazoknál használj SQL-szintű paginációt (OFFSET-FETCH) kliens oldali skip helyett, mert az hatékonyabb.

Diagnosztikai üzenetek

Hozzáférés a cursor.messages címekhez

Az messages attribútum tárolja az SQL utasítás végrehajtása során generált információs üzeneteket, ahogy azt a PEP 249 leírja. Ezek az üzenetek a PRINT utasítások és a 11 alatti súlyossági szinttel rendelkező RAISERROR kimenetét foglalják magukban.

Az attribútumok egy listája, ahol minden tuple tartalmaz egy üzenettípuskódot és az üzenet szövegét:

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

Az üzenet szövege illesztőprogram-előtaggal kapcsolatos információt tartalmaz, mert az illesztőprogram diagnosztikai rekordokként kéri le az üzeneteket a(z) SQLGetDiagRec használatával.

Üzenetek rögzítése tárolt eljárásokból

A végrehajtás után olvassa be a(z) cursor.messages elemet, hogy lekérje az előző utasításból származó PRINT kimenetet vagy a kiszolgáló tájékoztató üzeneteit:

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

Memóriakezelés

Nagy eredményeket hatékonyan dolgozzunk fel

Csomagokban fetchmany() kell letölteni azokat a táblákat, amelyek túl nagyok ahhoz, hogy egyszerre betöltsék a memóriába:

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

Generátor-alapú feldolgozás

Csomagoljuk be a batch lekérést egy generátorban, hogy soronként dolgozzon fel, miközben a memóriahasználat állandó marad az eredményhalmaz méretétől függetlenül:

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

Gyorsan zárd le a kurzorokat

Mindig zárd be a kurzorokat egy finally blokkoban, hogy szerveroldali erőforrásokat szabadítsanak fel, még akkor is, ha kivétel előfordul:

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

A kurzor állapotának kezelése

Ellenőrizze, hogy a kurzor tartalmaz-e adatot

Tesztelje, hogy egy lekérdezés visszaad-e bármilyen sort, ellenőrizve, hogy fetchone() visszaadja Nonea :

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

A kurzorok újrafelhasználása

Egy kurzor több lekérdezést is képes egymás után végrehajtani. Minden execute() hívás helyettesíti az előző eredményhalmazt:

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

Bevált gyakorlatok

Minta: Kurzorsegédosztály

Szervezd a kurzor életciklusának kezelését egy segédosztályba, hogy csökkentsd az ismétlődő kódot az alkalmazás egészében:

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 hagyd nyitva a kurzorokat

Egy kurzor, amely nincs kifejezetten zárva, a szerveroldali erőforrásokat tartja, amíg a kapcsolat le nem zárul. A tisztítás biztosításához használja a(z) try/finally elemet:

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

A kurzor élettartamának összeegyeztetése a működéshez

Készíts rövid életű kurzorokat egyetlen művelethez. Ugyanazt a kurzort csak kapcsolódó műveletek sorozatához használjuk újra:

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

Kapcsolódó tartalom

• Lekérdezések végrehajtása
• Adatok visszakeresése
• Kapcsolatkészletezés
• Tördelés