Not
Bu sayfaya erişim yetkilendirme gerektiriyor. Oturum açmayı veya dizinleri değiştirmeyi deneyebilirsiniz.
Bu sayfaya erişim yetkilendirme gerektiriyor. Dizinleri değiştirmeyi deneyebilirsiniz.
mssql-python sürücüsü, sorguları çalıştırmak, birden fazla sonuç kümesini işlemek ve belleği verimli yönetmek için imleç nesneleri sağlar.
İmleç temelleri
İmleç oluşturun ve kullanın
İmleç oluşturmak için conn.cursor() çağrısı yapın, ardından sorguları çalıştırmak ve sonuçları almak için execute() ve fetch yöntemlerini kullanın:
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()
Bağlam yöneticisi örüntüsü
Otomatik temizleme için bir bağlam yöneticisi uygulamak üzere bu with ifadeyi kullanın:
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
Birden çok imleç
Important
mssql-python sürücüsü, Çoklu Aktif Sonuç Kümeleri (MARS) desteğini sağlamaz. Tek bir bağlantıda birden fazla imleç oluşturabilirsiniz, ancak aynı anda sadece bir imleç aktif sorguya sahip olabilir. Her zaman tüm sonuçları bir imlecten getirin, sonra aynı bağlantıda başka bir imleçte çalıştırın.
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()
Sorguları eşzamanlı çalıştırmanız gerekiyorsa, ayrı bağlantılar kullanın:
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()
Getirme stratejileri
Tümünü getirme vs yinelemeli getirme
fetchall() kullanarak sonuç kümesinin tamamını tek seferde belleğe yükleyebilir veya satırları arabelleğe almadan teker teker işlemek için imleç üzerinde yineleme yapabilirsin.
# 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")
Gruplar halinde getir
Büyük sonuç kümelerini her şeyi belleğe yüklemeden parça halinde işlemek için toplu boyut ile fetchmany() kullanın.
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")
Tek değerler için fetchval kullanın
Tek bir değer döndüren skaler sorgular için kullanılır fetchval() . İlk satırın ilk sütununu döndürür.
# 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()
Birden fazla sonuç kümesi
Çoklu sonuç kümelerini işle
Önceki sonuç kümesindeki tüm satırları getirdikten sonra, geçerli sonuç kümesini atlayıp bir sonrakine ilerlemek için nextset() kullanılır.
# 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}")
Tüm sonuç kümelerini yineleme
nextset(), tek bir execute çağrısından tüm sonuç kümelerini tüketmek için False döndürene kadar döngüye alın:
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")
Daha fazla sonuç kümesi var mı kontrol edin
Kaç sonuç kümesi olduğunu önceden bilmeden tümünü işlemek için, bir döngü içinde nextset() öğesinin döndürdüğü değeri denetleyin:
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
İmleç tanımı
Sütun meta verilerine erişin
Bir sorgu yürütüldükten sonra, cursor.description içinde isim, tür kodu, görüntü boyutu, iç boyut, kesinlik, ölçek ve null olabilirlik durumunu içeren 7 öğeli demetlerden oluşan bir dizi — sütun başına bir tane — bulunur:
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)
Dinamik sonuç işleyicileri oluşturun
Çalışma zamanında sütun listesini cursor.description'dan oluşturarak herhangi bir sorguyla çalışan sonuç işleyicileri oluşturun:
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"])
Sonuç almayan sorguları yönet
cursor.description, INSERT, UPDATE ve DELETE gibi SELECT olmayan ifadelerden sonra None. Fetch yöntemlerini çağırmadan önce kontrol edin:
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()
Satır sayısı
Etkilenen satırları izleyin
INSERT, UPDATE veya DELETE sonrasında, cursor.rowcount ifadenin etkilediği satır sayısını döndürür:
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")
Bilinmeyen satır sayısını işle
# 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")
Satırları atla
Sayfa değiştirme alternatifi olarak skip kullanın
cursor.skip() satır getirmeden imlecin konumunu ilerletir. Büyük veri setleri için, daha iyi performans için SQL seviyesinde OFFSET-FETCH sayfa belirleme tercih edin:
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
Büyük veri kümeleri için, istemci tarafındaki skip işlemi yerine SQL düzeyinde sayfalama (OFFSET-FETCH) kullanın, çünkü daha verimlidir.
Tanı mesajları
cursor.messages'e erişim
Bu öznitelik messages , PEP 249'da tanımlandığı gibi SQL özeti çalıştırma sırasında oluşturulan bilgilendirici mesajları depolar. Bu mesajlar, PRINT deyimlerinden ve önem derecesi 11'in altında olan RAISERROR ifadelerinden gelen çıktıları içerir.
Öznitelik, her bir demetin bir ileti türü kodu ve ileti metni içerdiği bir demet listesidir:
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!')]
Mesaj metni, sürücü iletileri SQLGetDiagRec aracılığıyla tanılama kayıtları olarak aldığından sürücü öneki bilgisini içerir.
Depolanmış prosedürlerden mesaj yakalama
Önceki ifadeden herhangi bir PRINT çıktısını veya bilgi amaçlı sunucu iletilerini almak için yürütmenin ardından cursor.messages okuyun:
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}")
Bellek yönetimi
Büyük sonuçları verimli şekilde işlemek
Tek seferde belleğe yüklenemeyecek kadar büyük tabloları işlemek için fetchmany() kullanarak toplu halde getirin:
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
Jeneratör tabanlı işleme
Her seferinde bir satır işlemek için toplu veri getirmeyi bir jeneratör içine alın ve sonuç kümesinin boyutundan bağımsız olarak bellek kullanımını sabit tutun:
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
İmleci hemen kapatın
İstisna olsa bile sunucu tarafı kaynakları serbest bırakmak için bloktaki finally imleçleri her zaman kapatın:
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()
İmleç durumu yönetimi
İmlecin veri olup olmadığını kontrol edin
fetchone() ifadesinin None döndürüp döndürmediğini denetleyerek bir sorgunun herhangi bir satır döndürüp döndürmediğini test edin:
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}")
İmleçleri yeniden kullanın
Tek bir imleç, birden fazla sorguyu ardışık olarak çalıştırabilir. Her execute() çağrı önceki sonuç kümesini değiştirir:
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()
En iyi uygulamalar
Pattern: İmleç yardımcı sınıfı
İmleç yaşam döngüsü yönetimini, uygulamanız genelinde tekrarlayan kodları azaltmak için yardımcı bir sınıf içinde kapsülleştirin:
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")
İmleçleri açık bırakma
Açıkça kapalı olmayan bir imleç, bağlantı kapanana kadar sunucu tarafı kaynakları tutar. Temizliği garanti etmek için kullanın try/finally :
# 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()
İmleç ömrünü operasyonla eşleştirin
Tek işlemler için kısa ömürlü imleçler oluşturun. Aynı imleci yalnızca ilgili işlemler dizisi için tekrar kullanın:
# 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()