Notitie
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen u aan te melden of de directory te wijzigen.
Voor toegang tot deze pagina is autorisatie vereist. U kunt proberen de mappen te wijzigen.
De mssql-python-driver biedt cursorobjecten voor het uitvoeren van queries, het verwerken van meerdere resultaatsets en het efficiënt beheren van het geheugen.
Basisprincipes van cursor
Maak en gebruik cursors
Roep conn.cursor() aan om een cursor te maken, gebruik vervolgens en haal execute() methoden om queries uit te voeren en resultaten op te halen:
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()
Contextmanagerpatroon
Gebruik de with instructie om een contextmanager te implementeren voor automatische opruiming:
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
Meerdere cursors
Important
De mssql-python-driver ondersteunt geen Multiple Active Result Sets (MARS). Je kunt meerdere cursors op één verbinding maken, maar slechts één cursor kan tegelijk een actieve query hebben. Haal altijd alle resultaten van een cursor voordat je deze uitvoert op een andere cursor op dezelfde verbinding.
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()
Als je queries gelijktijdig moet uitvoeren, gebruik dan aparte verbindingen:
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()
Ophaalstrategieën
Alles ophalen versus iteratief ophalen
Gebruik fetchall() om de volledige resultaatset tegelijk in het geheugen te laden, of om over de cursor te itereren om rijen één voor één te verwerken zonder bufferen.
# 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")
In batches ophalen
Gebruik fetchmany() met batchgrootte om grote resultatensets in chunks te verwerken zonder alles in het geheugen te 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")
Gebruik fetchval voor enkele waarden
Gebruik fetchval() dit voor scalaire queries die één enkele waarde teruggeven. Het geeft de eerste kolom van de eerste rij terug.
# 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()
Meerdere resultaatsets
Verwerk meerdere resultaatsets
Gebruik nextset() om voorbij de huidige resultaatset te gaan naar de volgende nadat alle rijen uit de vorige set zijn opgehaald.
# 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}")
Doorloop alle resultaatsets
Herhaal totdat nextset()False retourneert om alle resultaatsets van één execute-aanroep te verwerken:
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")
Controleer of er meer resultaatsets bestaan
Controleer de retourwaarde van nextset() in een lus om alle resultatensets te verwerken zonder van tevoren te weten hoeveel er zijn:
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
Cursorbeschrijving
Toegang krijgen tot kolommetadata
Na het uitvoeren van een query cursor.description bevat het een reeks van tuples van 7 items — één per kolom — met naam, typecode, weergavegrootte, interne grootte, precisie, schaal en nullability:
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)
Bouw dynamische resultaathandlers
Bouw resultaathandlers die met elke query werken door de kolomlijst te construeren vanaf cursor.description tijdens runtime:
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"])
Verwerk query's zonder resultaat
cursor.description is None na niet-SELECT-uitspraken zoals INSERT, UPDATE, en DELETE. Controleer het voordat je fetch-methoden aanroept:
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()
Aantal rijen
Bijhouden van beïnvloede rijen
Na INSERT, UPDATE of DELETE retourneert cursor.rowcount het aantal rijen dat door de instructie wordt beïnvloed:
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")
Onbekend aantal rijen verwerken
# 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")
Rijen overslaan
Gebruik overslaan voor een alternatief voor paginering
cursor.skip() de cursorpositie verplaatst zonder rijen op te halen. Voor grote datasets geef je de voorkeur aan SQL-niveau OFFSET-FETCH paginering voor betere prestaties:
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)
Opmerking
Voor grote datasets gebruik je SQL-niveau paginering (OFFSET-FETCH) in plaats van client-side skip, omdat dat efficiënter is.
Diagnostische berichten
Toegang tot cursor.messages
Het messages attribuut slaat informatieberichten op die worden gegenereerd tijdens de uitvoering van SQL-instructies, zoals beschreven in PEP 249. Deze berichten bevatten output van PRINT verklaringen en RAISERROR met ernstniveaus onder 11.
Het attribuut is een lijst van tuples waarbij elke tuple een berichttypecode en de berichttekst bevat:
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!')]
De tekst van het bericht bevat informatie over het voorvoegsel van de driver, omdat de driver berichten ophaalt in de vorm van diagnostische records via SQLGetDiagRec.
Berichten vastleggen uit opgeslagen procedures
Lees cursor.messages na uitvoering om eventuele PRINT-uitvoer of informatieberichten van de server van de vorige instructie op te halen:
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}")
Geheugenbeheer
Verwerk grote resultaten efficiënt
Haal gegevens in batches op met fetchmany() om tabellen te verwerken die te groot zijn om in één keer in het geheugen te laden:
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
Generatorgebaseerde verwerking
Verpak het batchgewijs ophalen in een generator om één rij tegelijk te verwerken, terwijl het geheugengebruik constant blijft, ongeacht de grootte van de resultaatset:
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
Sluit cursoren onmiddellijk
Sluit altijd cursors in een finally blok om server-side resources vrij te geven, zelfs als er een uitzondering optreedt:
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()
Cursorstatusbeheer
Controleer of de cursor data heeft
Controleer of een query rijen heeft geretourneerd door te controleren of fetchone()None retourneert:
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}")
Cursors hergebruiken
Een enkele cursor kan meerdere queries achter elkaar uitvoeren. Elke execute() oproep vervangt de vorige resultaatset:
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()
Beste praktijken
Patroon: Cursorhelper-klasse
Breng het beheer van de levenscyclus van de cursor onder in een helperklasse om de hoeveelheid standaardcode in je hele applicatie te verminderen:
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")
Laat de cursors niet open
Een cursor die niet expliciet gesloten is, houdt server-side resources vast totdat de verbinding wordt gesloten. Gebruik try/finally om schoonmaak te garanderen:
# 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()
Stem de levensduur van de cursor af op de werking
Maak kortstondige cursors voor enkele bewerkingen. Hergebruik dezelfde cursor alleen voor een reeks gerelateerde bewerkingen:
# 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()