Kommentar
Åtkomst till den här sidan kräver auktorisering. Du kan prova att logga in eller ändra kataloger.
Åtkomst till den här sidan kräver auktorisering. Du kan prova att ändra kataloger.
mssql-python-drivrutinen tillhandahåller markörobjekt för att köra frågor, hantera flera resultatuppsättningar och hantera minnet effektivt.
Grunderna om markören
Skapa och använd markörer
Anropa conn.cursor() för att skapa en markör och använd sedan execute() och hämtningsmetoder för att köra frågor och hämta resultat:
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()
Kontexthanterarmönster
Använd satsen with för att implementera en kontexthanterare för automatisk rensning:
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
Flera markörer
Important
mssql-python-drivrutinen stöder inte Multiple Active Result Sets (MARS). Du kan skapa flera markörer på en och samma anslutning, men bara en markör kan ha en aktiv fråga åt gången. Hämta alltid alla resultat från en markör innan du kör på en annan markör på samma anslutning.
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()
Om du behöver köra frågor samtidigt, använd separata anslutningar istället:
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()
Hämtastrategier
Hämtning av allt vs iterativ hämtning
Använd fetchall() för att ladda hela resultatuppsättningen i minnet på en gång, eller iterera över markören för att bearbeta rader en i taget utan buffring.
# 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")
Hämta i omgångar
Använd fetchmany() med batchstorlek för att bearbeta stora resultatuppsättningar i bitar utan att ladda in allt i minnet.
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")
Använd fetchval för enskilda värden
Använd fetchval() för skalärfrågor som returnerar ett enda värde. Den returnerar den första kolumnen i första raden.
# 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()
Flera resultatuppsättningar
Bearbeta flera resultatmängder
Använd nextset() för att gå förbi den aktuella resultatuppsättningen till nästa efter att ha hämtat alla rader från föregående uppsättning.
# 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}")
Iterera alla resultatmängder
Loopa tills nextset() returnerar False för att bearbeta alla resultatuppsättningar från ett enda anrop till 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")
Kontrollera om fler resultatmängder finns
Kontrollera returvärdet för nextset() i en loop för att bearbeta alla resultatmängder utan att i förväg veta hur många det finns:
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
Beskrivning av markören
Få åtkomst till kolumnmetadata
Efter att ha kört en fråga cursor.description innehåller den en sekvens av tupler med 7 objekt — en per kolumn — med namn, typkod, visningsstorlek, intern storlek, precision, skala och nullbarhet:
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)
Bygg dynamiska resultathanterare
Bygg resultathanterare som fungerar med valfri fråga genom att konstruera kolumnlistan från cursor.description vid körning:
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"])
Hantera frågor utan resultat
cursor.description är None efter icke-SELECT-satser såsom INSERT, UPDATE, och DELETE. Kontrollera detta innan du anropar fetch-metoder:
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()
Antal rader
Spåra påverkade rader
Efter INSERT, UPDATE, eller DELETE, cursor.rowcount returnerar antalet rader som påverkas av påståendet:
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")
Hantera okänt antal rader
# 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")
Hoppa över rader
Använd hoppa för ett alternativ till paginering
cursor.skip() avancerar markörpositionen utan att hämta rader. För stora datamängder, föredra SQL-nivå OFFSET-FETCH paginering för bättre prestanda:
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)
Note
För stora datamängder bör du använda paginering på SQL-nivå (OFFSET-FETCH) i stället för överhoppning på klientsidan, eftersom det är mer effektivt.
Diagnostiska meddelanden
Öppna cursor.messages
Attributet messages lagrar informationsmeddelanden som genereras under SQL-satsexekveringen, enligt PEP 249. Dessa meddelanden omfattar utdata från PRINT-satser och från RAISERROR med allvarlighetsnivåer under 11.
Attributet är en lista över tupler där varje tuple innehåller en meddelandetypkod och meddelandetexten:
conn = mssql_python.connect(connection_string, autocommit=True)
cursor = conn.cursor()
cursor.execute("PRINT 'Hello world!'")
print(cursor.messages)
Resultat:
[('[01000] (0)', '[Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Hello world!')]
Meddelandetexten innehåller drivrutinsprefixinformation eftersom drivrutinen hämtar meddelanden som diagnostiska poster genom SQLGetDiagRec.
Fånga meddelanden från lagrade procedurer
Läs cursor.messages efter körningen för att hämta eventuell utdata från PRINT eller informerande servermeddelanden från den föregående instruktionen:
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}")
Minneshantering
Bearbeta stora resultat effektivt
Hämta batchvis med fetchmany() för att bearbeta tabeller som är för stora för att läsa in i minnet på en gång:
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
Generatorbaserad bearbetning
Lägg batchhämtning i en generator för att bearbeta en rad i taget samtidigt som minnesanvändningen hålls konstant oavsett storleken på resultatuppsättningen:
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
Stäng markörerna snabbt
Stäng alltid markörer i ett finally block för att frigöra serverresurser även om ett undantag inträffar:
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()
Styr av markörens tillstånd
Kontrollera om markören har data
Testa om en fråga returnerade några rader genom att kontrollera om fetchone() returnerar 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}")
Återanvänd markörer
En enda markör kan utföra flera frågor i följd. Varje execute() anrop ersätter den föregående resultatuppsättningen:
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()
Metodtips
Mönster: Markörhjälpklass
Kapsla in kursorens livscykelhantering i en hjälpklass för att minska boilerplate i din applikation:
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")
Lämna inte markörerna öppna
En markör som inte är uttryckligen stängd håller serverresurser tills anslutningen stängs. Använd try/finally för att garantera städning:
# 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()
Matcha markörens livslängd till operationen
Skapa kortlivade markörer för enskilda operationer. Återanvänd samma markör endast för en sekvens av relaterade operationer:
# 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()