Gestionar cursores y conjuntos de resultados

El controlador mssql-python proporciona objetos cursor para ejecutar consultas, gestionar múltiples conjuntos de resultados y gestionar la memoria de forma eficiente.

Conceptos básicos del cursor

Crear y usar cursores

Llama conn.cursor() para crear un cursor, luego usa execute() y obtén métodos para ejecutar consultas y obtener 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()

Patrón de gestor de contexto

Utiliza la with instrucción para implementar un gestor de contexto para la limpieza 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

Varios cursores

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()

Si necesitas hacer consultas simultáneamente, usa conexiones separadas en su lugar:

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()

Estrategias de búsqueda

Obtener todo frente a obtención iterativa

Usa fetchall() para cargar todo el conjunto de resultados en memoria de una vez, o itera sobre el cursor para procesar las filas una a una sin almacenamiento en búfer.

# 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")

Recoger en lotes

Úsalo fetchmany() con un tamaño de lote para procesar grandes conjuntos de resultados en bloques sin cargar todo en la memoria.

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")

Usa fetchval para valores individuales

Úsase fetchval() para consultas escalares que devuelvan un solo valor. Devuelve la primera columna de la primera fila.

# 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()

Varios conjuntos de resultados

Procesar varios conjuntos de resultados

Úsalo nextset() para avanzar más allá del conjunto de resultados actual al siguiente después de recuperar todas las filas del 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 los conjuntos de resultados

Repita hasta que nextset() devuelva False para consumir todos los conjuntos de resultados en una sola llamada 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")

Comprueba si existen más conjuntos de resultados

Comprueba el valor de retorno de nextset() en un bucle para consumir todos los conjuntos de resultados sin saber de antemano cuántos hay:

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

Descripción del cursor

Metadatos de la columna de acceso

Tras ejecutar una consulta, cursor.description contiene una secuencia de tuplas de 7 elementos —una por columna— con el nombre, el código de tipo, el tamaño de visualización, el tamaño interno, la precisión, la escala y la nulabilidad:

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 manejadores dinámicos de resultados

Construye manejadores de resultados que funcionen con cualquier consulta construyendo la lista de columnas desde cursor.description en tiempo de ejecución:

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"])

Gestionar consultas sin resultados

cursor.description es None después de sentencias no SELECT como INSERT, UPDATE, y DELETE. Compruébalo antes de llamar a los 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()

Recuento de filas

Rastrear filas afectadas

Después de INSERT, UPDATE, o DELETE, cursor.rowcount devuelve el número de filas afectadas por la afirmación:

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")

Gestionar un número desconocido de filas

# 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")

Omitir filas

Usa skip como alternativa a la paginación

cursor.skip() avanza la posición del cursor sin recuperar filas. Para conjuntos de datos grandes, prefiera la paginación OFFSET-FETCH a nivel SQL para un mejor rendimiento:

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)

Mensajes diagnósticos

Mensajería cursor de acceso

El messages atributo almacena mensajes informativos generados durante la ejecución de la sentencia SQL, tal y como se describe en PEP 249. Estos mensajes incluyen la salida de las instrucciones PRINT y RAISERROR con niveles de gravedad inferiores a 11.

El atributo es una lista de tuplas donde cada tupla contiene un código de tipo de mensaje y el texto del mensaje:

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!')]

El texto del mensaje incluye la información del prefijo del controlador porque el controlador recupera los mensajes como registros diagnósticos a través de SQLGetDiagRec.

Captura de mensajes desde procedimientos almacenados

Lea cursor.messages después de ejecutar para recuperar cualquier salida de PRINT o mensaje informativo del servidor de la instrucción 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}")

Administración de memoria

Procesar resultados grandes de forma eficiente

Recupere por lotes usando fetchmany() para procesar tablas demasiado grandes para cargar en memoria de una sola 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

Procesamiento basado en generadores

Envuelva la obtención por lotes en un generador para procesar una fila a la vez manteniendo el uso de memoria constante independientemente del tamaño del 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

Cierra los cursores de inmediato

Cierra siempre los cursores en un finally bloque para liberar recursos del lado del servidor incluso si ocurre una excepció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()

Gestión del estado del cursor

Comprueba si el cursor tiene datos

Comprobar si una consulta devuelve alguna fila comprobando si fetchone() devuelve 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

Un solo cursor puede ejecutar varias consultas de forma secuencial. Cada execute() llamada reemplaza al 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()

procedimientos recomendados

Patrón: Clase auxiliar de cursor

Encapsula la gestión del ciclo de vida del cursor en una clase auxiliar para reducir el código estándar en toda tu aplicación:

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")

No dejes los cursores abiertos

Un cursor que no está explícitamente cerrado almacena recursos del lado del servidor hasta que se cierra la conexión. Usa try/finally para garantizar la limpieza:

# 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()

Ajusta la duración del cursor a la operación

Crear cursores de corta duración para operaciones individuales. Reutiliza el mismo cursor solo para una secuencia de operaciones 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()

Contenido relacionado

• Ejecución de consultas
• Recuperación de datos
• Agrupación de conexiones
• Paginación