Catatan
Akses ke halaman ini memerlukan otorisasi. Anda dapat mencoba masuk atau mengubah direktori.
Akses ke halaman ini memerlukan otorisasi. Anda dapat mencoba mengubah direktori.
Driver mssql-python menyediakan objek kursor untuk mengeksekusi kueri, menangani beberapa set hasil, dan mengelola memori secara efisien.
Dasar-dasar kursor
Membuat dan menggunakan kursor
Panggil conn.cursor() untuk membuat kursor, lalu gunakan execute() dan ambil metode untuk menjalankan kueri dan mengambil hasil:
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()
Pola pengelola konteks
Gunakan pernyataan with untuk mengimplementasikan manajer konteks untuk pembersihan otomatis:
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
Beberapa kursor
Penting
Driver mssql-python tidak mendukung Beberapa Set Hasil Aktif (MARS). Anda dapat membuat beberapa kursor pada satu koneksi, tetapi hanya satu kursor yang dapat memiliki kueri aktif pada satu waktu. Selalu ambil semua hasil dari kursor sebelum dijalankan pada kursor lain pada koneksi yang sama.
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()
Jika Anda perlu menjalankan kueri secara bersamaan, gunakan koneksi terpisah sebagai gantinya:
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()
Strategi pengambilan data
Ambil semua vs pengambilan berulang
Gunakan fetchall() untuk memuat seluruh kumpulan hasil ke dalam memori sekaligus, atau iterasi kursor untuk memproses baris satu per satu tanpa buffering.
# 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")
Ambil dalam batch
Gunakan fetchmany() dengan ukuran batch untuk memproses kumpulan hasil besar dalam potongan tanpa memuat semuanya ke dalam memori.
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")
Gunakan fetchval untuk nilai tunggal
Gunakan fetchval() untuk kueri skalar yang mengembalikan satu nilai. Mengembalikan kolom pertama dari baris pertama.
# 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()
Beberapa kumpulan hasil
Memproses beberapa set hasil
Gunakan nextset() untuk beralih dari kumpulan hasil saat ini ke kumpulan hasil berikutnya setelah mengambil semua baris dari kumpulan hasil sebelumnya.
# 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}")
Ulangi semua set hasil
Ulangi hingga nextset() mengembalikan False untuk memproses semua kumpulan hasil dari satu kali pemanggilan eksekusi:
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")
Periksa apakah ada lebih banyak kumpulan hasil
Periksa nilai yang dikembalikan oleh nextset() dalam sebuah perulangan untuk memproses semua kumpulan hasil tanpa mengetahui terlebih dahulu ada berapa banyak.
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
Deskripsi kursor
Akses metadata kolom
Setelah menjalankan kueri, cursor.description berisi urutan tuple berisi 7 item — satu per kolom — yang mencakup nama, kode jenis, ukuran tampilan, ukuran internal, presisi, skala, dan apakah dapat bernilai null:
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)
Membangun penanganan hasil dinamis
Buat handler hasil yang dapat bekerja dengan kueri apa pun dengan menyusun daftar kolom dari cursor.description saat 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"])
Menangani kueri tanpa hasil
cursor.description adalah None setelah pernyataan non-SELECT seperti INSERT, UPDATE, dan DELETE. Periksa sebelum memanggil metode fetch:
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()
Jumlah baris
Lacak baris yang terpengaruh
Setelah INSERT, UPDATE, atau DELETE, cursor.rowcount mengembalikan jumlah baris yang terpengaruh oleh pernyataan:
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")
Menangani jumlah baris yang tidak diketahui
# 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")
Lewati baris
Gunakan lewati untuk alternatif penomoran halaman
cursor.skip() menggeser posisi kursor tanpa mengambil data baris. Untuk kumpulan data besar, sebaiknya gunakan paginasi OFFSET-FETCH pada level SQL untuk performa yang lebih baik:
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
Untuk kumpulan data besar, gunakan paginasi pada level SQL (OFFSET-FETCH) alih-alih skip di sisi klien, karena lebih efisien.
Pesan diagnostik
Akses cursor.messages
Atribut menyimpan messages pesan informasi yang dihasilkan selama eksekusi pernyataan SQL, seperti yang dijelaskan dalam PEP 249. Pesan-pesan ini mencakup keluaran dari pernyataan PRINT dan RAISERROR dengan tingkat keparahan di bawah 11.
Atribut adalah daftar tuple di mana setiap tuple berisi kode jenis pesan dan teks pesan:
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!')]
Teks pesan menyertakan informasi awalan driver karena driver mengambil pesan sebagai catatan diagnostik melalui SQLGetDiagRec.
Menangkap pesan dari prosedur tersimpan
Baca cursor.messages setelah eksekusi untuk mengambil output PRINT atau pesan informatif server apa pun dari pernyataan sebelumnya:
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}")
Manajemen memori
Memproses hasil besar secara efisien
Ambil dalam batch menggunakan fetchmany() untuk memproses tabel yang terlalu besar untuk dimuat ke memori sekaligus:
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
Pemrosesan berbasis generator
Bungkus pengambilan data secara batch dalam generator untuk memproses satu baris pada satu waktu sehingga penggunaan memori tetap konstan terlepas dari ukuran kumpulan hasil:
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
Tutup kursor segera
Selalu tutup kursor dalam finally blok untuk melepaskan sumber daya sisi server meskipun terjadi pengecualian:
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()
Manajemen status kursor
Memeriksa apakah kursor memiliki data
Uji apakah kueri mengembalikan baris apa pun dengan memeriksa apakah fetchone() mengembalikan 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}")
Gunakan kembali kursor
Satu kursor dapat mengeksekusi beberapa kueri secara berurutan. Setiap execute() panggilan menggantikan kumpulan hasil sebelumnya:
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()
Praktik terbaik
Pola: Kelas pembantu kursor
Merangkum manajemen siklus hidup kursor dalam kelas pembantu untuk mengurangi boilerplate di seluruh aplikasi Anda:
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")
Jangan biarkan kursor terbuka
Kursor yang tidak ditutup secara eksplisit menyimpan sumber daya sisi server hingga koneksi ditutup. Gunakan try/finally untuk menjamin pembersihan:
# 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()
Cocokkan masa pakai kursor dengan operasi
Buat kursor berumur pendek untuk operasi tunggal. Gunakan kembali kursor yang sama hanya untuk urutan operasi terkait:
# 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()