Observação
O acesso a essa página exige autorização. Você pode tentar entrar ou alterar diretórios.
O acesso a essa página exige autorização. Você pode tentar alterar os diretórios.
O driver mssql-python fornece objetos cursor para execução de consultas, gerenciamento de múltiplos conjuntos de resultados e gerenciamento eficiente da memória.
Fundamentos do cursor
Criar e usar cursores
Chame conn.cursor() para criar um cursor, depois use execute() e busque métodos para rodar consultas e recuperar resultados:
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()
Padrão de gerenciador de contexto
Use a with instrução para implementar um gerenciador de contexto para limpeza automática:
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
Vários cursores
Importante
O driver mssql-python não suporta Múltiplos Conjuntos de Resultados Ativos (MARS). Você pode criar múltiplos cursores em uma única conexão, mas apenas um cursor pode ter uma consulta ativa ao mesmo tempo. Sempre busque todos os resultados de um cursor antes de executar em outro cursor na mesma conexão.
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()
Se você precisar rodar consultas simultaneamente, use conexões separadas:
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()
Estratégias de busca
Busca completa vs. busca iterativa
Use fetchall() para carregar todo o conjunto de resultados na memória de uma só vez ou itere sobre o cursor para processar as linhas uma de cada vez, sem armazenamento em buffer.
# 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")
Buscar em lotes
Use fetchmany() com tamanho de lote para processar grandes conjuntos de resultados em blocos sem carregar tudo na memória.
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")
Use o fetchval para valores individuais
Use fetchval() para consultas escalares que retornam um único valor. Retorna a primeira coluna da primeira linha.
# 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()
Vários conjuntos de resultados
Processar vários conjuntos de resultados
Use nextset() para avançar além do conjunto de resultados atual para o próximo após buscar todas as linhas do conjunto anterior.
# 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}")
Iterar todos os conjuntos de resultados
Repita até que nextset() retorne False para consumir todos os conjuntos de resultados de uma única chamada a 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")
Verifique se existem mais conjuntos de resultados
Verifique o valor de retorno de nextset() em um ciclo para consumir todos os conjuntos de resultados sem saber antecipadamente quantos existem:
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
Descrição do cursor
Acesse os metadados da coluna
Após executar uma consulta, cursor.description contém uma sequência de tuplas de 7 itens — uma por coluna — com nome, código de tipo, tamanho de exibição, tamanho interno, precisão, escala e anulabilidade:
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)
Construir manipuladores dinâmicos de resultados
Construa manipuladores de resultados que funcionem com qualquer consulta construindo a lista de colunas a partir de cursor.description em tempo de execução:
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"])
Lidar com consultas sem resultados
cursor.description é None após instruções não-SELECT como INSERT, UPDATE, e DELETE. Verifique isso antes de chamar os métodos 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()
Contagem de linhas
Acompanhar linhas afetadas
Após INSERT, UPDATE, ou DELETE, cursor.rowcount retorna o número de linhas afetadas pela afirmação:
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")
Lidar com contagem de linhas desconhecida
# 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")
Pular linhas
Use skip para alternativa de paginação
cursor.skip() avança a posição do cursor sem buscar linhas. Para grandes conjuntos de dados, prefira paginação OFFSET-FETCH em nível SQL para melhor desempenho:
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
Para conjuntos de dados grandes, use paginação em nível de SQL (OFFSET-FETCH) em vez de usar skip no cliente, pois é mais eficiente.
Mensagens de diagnóstico
Mensagens de cursor de acesso
O messages atributo armazena mensagens informativas geradas durante a execução da instrução SQL, conforme descrito no PEP 249. Essas mensagens incluem a saída das instruções PRINT e de RAISERROR com níveis de gravidade abaixo de 11.
O atributo é uma lista de tuplas onde cada tupla contém um código de tipo de mensagem e o texto da mensagem:
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!')]
O texto da mensagem inclui informações sobre o prefixo do driver porque o driver recupera mensagens como registros de diagnóstico por meio de SQLGetDiagRec.
Capturar mensagens a partir de procedimentos armazenados
Leia cursor.messages após a execução para recuperar quaisquer PRINT mensagens de saída ou informações do servidor da instrução anterior:
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}")
Gerenciamento de memória
Processe resultados grandes de forma eficiente
Busque em lotes usando fetchmany() para processar tabelas grandes demais para serem carregadas na memória de uma só vez:
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
Processamento baseado em geradores
Envolva a busca por lote em um gerador para processar uma linha de cada vez, mantendo o uso de memória constante independentemente do tamanho do conjunto de resultados:
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
Feche os cursores imediatamente
Sempre feche os cursores em um finally bloco para liberar recursos do lado do servidor, mesmo que ocorra uma exceção:
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()
Gerenciamento do estado do cursor
Verifique se o cursor tem dados
Teste se uma consulta retornou alguma linha verificando se fetchone() retorna 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}")
Reutilizar cursores
Um único cursor pode executar múltiplas consultas sequencialmente. Cada execute() chamada substitui o conjunto de resultados anterior:
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()
Práticas recomendadas
Padrão: Classe auxiliar de cursor
Encapsule o gerenciamento do ciclo de vida do cursor em uma classe auxiliar para reduzir o padrão padrão em toda a sua aplicação:
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")
Não deixe os cursores abertos
Um cursor que não está explicitamente fechado mantém recursos do lado do servidor até que a conexão seja fechada. Use try/finally para garantir a limpeza:
# 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()
Ajuste a vida útil do cursor à operação
Crie cursores de curta duração para operações únicas. Reuse o mesmo cursor apenas para uma sequência de operações relacionadas:
# 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()