Chamar procedimentos armazenados com mssql-python

O driver mssql-python não implementa o callproc() método da especificação DB-API 2.0. Chamar callproc() gera NotSupportedError. Em vez disso, use a sequência de escape ODBC {CALL ...} com métodos padrão de execução de consultas.

Execução básica de um procedimento armazenado

Sem parâmetros

Execute um procedimento armazenado usando a {CALL} sequência de escape. Este exemplo chama o sp_databases procedimento armazenado do sistema, que não aceita parâmetros e retorna uma linha por banco de dados:

import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("{CALL sp_databases}")

for row in cursor:
    print(row.DATABASE_NAME, row.DATABASE_SIZE)

Com parâmetros de entrada

Passe parâmetros usando espaços reservados posicionais ou nomeados:

# Single parameter
cursor.execute(
    "{CALL dbo.uspGetManagerEmployees(?)}", (16,)
)

for row in cursor:
    print(row.FirstName, row.LastName)

Parâmetros de saída

Declarar e recuperar parâmetros de saída

Os procedimentos armazenados do SQL Server podem retornar valores por meio de parâmetros de saída. Use variáveis Transact-SQL (T-SQL) para capturar os valores de saída e, em seguida, recupere-os com uma instrução SELECT:

cursor.execute("""
    DECLARE @total_out MONEY;
    SELECT @total_out = SUM(TotalDue)
    FROM Sales.SalesOrderHeader
    WHERE CustomerID = %(customer_id)s;
    SELECT @total_out AS TotalAmount;
""", {"customer_id": 29825})

row = cursor.fetchone()
total = row.TotalAmount
print(f"Customer total: ${total}")

Múltiplos parâmetros de saída

Capture múltiplos valores de saída declarando variáveis separadas. O mesmo padrão funciona com qualquer procedimento armazenado que tenha OUTPUT parâmetros:

cursor.execute("""
    DECLARE @total_orders INT, @total_spent MONEY;
    SELECT @total_orders = COUNT(*), @total_spent = SUM(TotalDue)
    FROM Sales.SalesOrderHeader
    WHERE CustomerID = %(cust_id)s;
    SELECT @total_orders AS OrderCount, @total_spent AS TotalSpent;
""", {"cust_id": 29825})

stats = cursor.fetchone()
print(f"Orders: {stats.OrderCount}, Total spent: ${stats.TotalSpent}")

Retornar valores

Capturar o valor de retorno do procedimento armazenado

Execute um procedimento armazenado que retorne resultados:

cursor.execute("EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = %(emp_id)s", {"emp_id": 5})

rows = cursor.fetchall()
if rows:
    print(f"Found {len(rows)} managers in chain")
    for row in rows:
        print(f"  Manager: {row.FirstName} {row.LastName}")
else:
    print("No managers found")

Valor de retorno com parâmetros de saída

cursor.execute("""
    DECLARE @return_value INT, @message NVARCHAR(500);
    SELECT @return_value = CASE WHEN COUNT(*) > 0 THEN 0 ELSE 1 END,
           @message = CASE WHEN COUNT(*) > 0 THEN N'Customer found' ELSE N'Customer not found' END
    FROM Sales.Customer WHERE CustomerID = %(cust_id)s;
    SELECT @return_value AS ReturnCode, @message AS Message;
""", {"cust_id": 29825})

result = cursor.fetchone()
print(f"Return code: {result.ReturnCode}, Message: {result.Message}")

Conjuntos de resultados

Conjunto único de resultados

Execute um procedimento armazenado que retorne um único conjunto de resultados:

cursor.execute(
    "{CALL dbo.uspGetBillOfMaterials(?, ?)}",
    (800, "2026-01-01")
)

customers = cursor.fetchall()
for row in customers:
    print(f"{row.ProductAssemblyID}: {row.ComponentDesc}")

Vários conjuntos de resultados

Para cada resultado para o qual você avança, nextset() atualiza rowcount. Um resultado de modificação de dados reporta sua própria contagem de linhas afetadas e tem description is None; não chame métodos de busca para esse resultado. Um resultado produtor de linhas começa com rowcount == -1 antes de buscar. SET NOCOUNT ON suprime os resultados das linhas afetadas. Para um lote que mistura esses tipos de resultados, veja Acompanhar cada resultado em um lote.

Alguns procedimentos armazenados retornam vários conjuntos de resultados. Use nextset() para navegar entre eles:

cursor.execute("""
    SELECT TOP 1 SalesOrderID, OrderDate, TotalDue
    FROM Sales.SalesOrderHeader WHERE CustomerID = 29825;
    SELECT TOP 3 Name, ListPrice
    FROM Production.Product WHERE ListPrice > 0
    ORDER BY ListPrice DESC;
""")

# First result set: order header
order = cursor.fetchone()
print(f"Order: {order.SalesOrderID}, Date: {order.OrderDate}")

cursor.nextset()

# Second result set: products
print("Products:")
for item in cursor:
    print(f"  {item.Name}: ${item.ListPrice}")

Verifique se há mais conjuntos de resultados

Percorrer todos os conjuntos de resultados retornados por um procedimento armazenado usando nextset():

cursor.execute("SELECT TOP 3 ProductID, Name FROM Production.Product; SELECT TOP 3 FirstName, LastName FROM Person.Person")

result_set_num = 1
while True:
    print(f"--- Result Set {result_set_num} ---")
    for row in cursor:
        print(row)
    
    if not cursor.nextset():
        break
    result_set_num += 1

Transações com procedimentos armazenados

Controle explícito de transações

Envolva múltiplas chamadas de procedimento armazenado em uma transação para garantir atomicidade:

conn.autocommit = False

try:
    cursor.execute("{CALL dbo.DebitAccount(?, ?)}", (1001, 100.00))
    
    cursor.execute("{CALL dbo.CreditAccount(?, ?)}", (1002, 100.00))
    
    conn.commit()
    print("Transfer completed")
except mssql_python.DatabaseError as e:
    conn.rollback()
    print(f"Transfer failed: {e}")

Deixe o procedimento armazenado gerenciar a transação

Se o procedimento armazenado lida com suas próprias transações:

conn.autocommit = True  # Let SP manage transactions

cursor.execute("""
    DECLARE @result INT;
    EXECUTE @result = dbo.TransferFunds 
        @FromAccount = %(from_acc)s,
        @ToAccount = %(to_acc)s,
        @Amount = %(amount)s;
    SELECT @result AS TransferResult;
""", {"from_acc": 1001, "to_acc": 1002, "amount": 100.00})

result = cursor.fetchone()
if result.TransferResult == 0:
    print("Transfer successful")

Tratamento de erros

Capturar erros de procedimentos armazenados

Gerencie exceções levantadas por procedimentos armazenados ou instruções Transact-SQL:

try:
    cursor.execute("{CALL dbo.DangerousProcedure}")
except mssql_python.ProgrammingError as e:
    # Handle SQL errors raised by RAISERROR or THROW
    print(f"Stored procedure error: {e}")
except mssql_python.DatabaseError as e:
    # Handle other database errors
    print(f"Database error: {e}")

Capturar declarações e mensagens informativas do PRINT

As instruções do SQL Server PRINT e RAISERROR com gravidade inferior a 11 são registradas em cursor.messages após a execução. Cada entrada é uma (message_type, message_text) tupla. Quando um PRINT é executado antes de um conjunto de resultados, ele ocupa um conjunto de resultados sem linhas próprio, então leia cursor.messages primeiro e depois chame nextset() para acessar as linhas:

cursor.execute("PRINT 'Operation complete'; SELECT 1 AS Status")

for msg_type, msg_text in cursor.messages:
    print(f"Server message: {msg_text}")

cursor.nextset()
row = cursor.fetchone()
print(f"Status: {row.Status}")

Quando um procedimento armazenado emite PRINT mensagens entre múltiplos conjuntos de resultados, leia cursor.messages após execute() e novamente após cada nextset() chamada para que mensagens de todos os conjuntos de resultados sejam capturadas:

cursor.execute("""
    PRINT 'Starting first result set';
    SELECT TOP 3 ProductID, Name FROM Production.Product;
    PRINT 'Starting second result set';
    SELECT TOP 3 FirstName, LastName FROM Person.Person;
""")

all_messages = []
while True:
    all_messages.extend(cursor.messages)
    if cursor.description:
        for row in cursor:
            print(row)
    if not cursor.nextset():
        break

for _, text in all_messages:
    print(f"Server: {text}")

Para a API completa cursor.messages , veja Gerenciamento de cursores.

Práticas recomendadas

Uso de parâmetros nomeados

Parâmetros nomeados são mais claros e não dependem da ordem:

# Recommended: {CALL} with positional parameters
cursor.execute(
    "{CALL dbo.uspGetBillOfMaterials(?, ?)}",
    (800, "2026-01-01")
)

# Also valid: EXECUTE with named T-SQL parameters
cursor.execute("""
    EXECUTE dbo.uspGetBillOfMaterials
        @StartProductID = ?,
        @CheckDate = ?
""", (800, "2026-01-01"))

Gerenciar parâmetros de saída anuláveis

Verifique valores NULL ao recuperar parâmetros de saída de procedimentos armazenados:

cursor.execute("""
    DECLARE @optional_value NVARCHAR(100);
    SELECT @optional_value = Color FROM Production.Product WHERE ProductID = %(id)s;
    SELECT @optional_value AS OutputValue;
""", {"id": 1})

result = cursor.fetchone()
if result.OutputValue is not None:
    print(f"Value: {result.OutputValue}")
else:
    print("No value returned")

Use SET NOCOUNT ON em procedimentos armazenados

Para melhor desempenho e tratamento de resultados mais limpo, certifique-se de que seus procedimentos armazenados incluam:

CREATE PROCEDURE dbo.MyProcedure
AS
BEGIN
    SET NOCOUNT ON;  -- Prevents "n rows affected" messages
    -- procedure logic
END

Exemplo: Fluxo de trabalho completo

Aqui está um exemplo prático que chama um procedimento armazenado, recupera valores de saída e lida com erros:

import mssql_python

def get_employee_report(manager_id: int) -> dict:
    """Look up a manager's employees and compute average vacation hours."""
    conn = mssql_python.connect(connection_string)
    conn.autocommit = False
    cursor = conn.cursor()

    try:
        # Get manager info
        cursor.execute("""
            SELECT BusinessEntityID, JobTitle
            FROM HumanResources.Employee
            WHERE BusinessEntityID = %(mgr)s
        """, {"mgr": manager_id})
        mgr = cursor.fetchone()
        print(f"Manager {mgr.BusinessEntityID}: {mgr.JobTitle}")

        # Get direct reports via stored procedure
        cursor.execute("{CALL dbo.uspGetManagerEmployees(?)}", (manager_id,))
        employees = cursor.fetchall()
        print(f"Found {len(employees)} employee(s)")

        # Compute average vacation hours
        cursor.execute("""
            DECLARE @avg_hours INT;
            SELECT @avg_hours = AVG(VacationHours)
            FROM HumanResources.Employee;
            SELECT @avg_hours AS AvgVacation;
        """)
        avg = cursor.fetchone().AvgVacation
        print(f"Avg vacation hours: {avg}")

        conn.commit()
        return {"manager": mgr.JobTitle, "reports": len(employees), "avg_vacation": avg}

    except mssql_python.DatabaseError as e:
        conn.rollback()
        raise
    finally:
        cursor.close()
        conn.close()

# Usage
result = get_employee_report(manager_id=16)

Obtenha chaves geradas usando OUTPUT INSERTED

Para recuperar um valor de identidade de uma inserção em INSERT (com ou sem procedimento armazenado), use OUTPUT INSERTED em vez de SCOPE_IDENTITY(). Essa abordagem retorna o valor no mesmo conjunto de resultados:

cursor.execute("""
    INSERT INTO Production.ProductCategory (Name)
    OUTPUT INSERTED.ProductCategoryID
    VALUES (%(name)s)
""", {"name": "Custom Parts"})

new_id = cursor.fetchval()
print(f"New category ID: {new_id}")

Esse padrão funciona para qualquer tabela com uma coluna identidade e não requer um procedimento armazenado.

Recursos sem suporte

callproc()

O driver mssql-python gera NotSupportedError se você chamar cursor.callproc(). Use cursor.execute("{CALL ...}") ou cursor.execute("EXECUTE ...") , em vez disso, conforme mostrado ao longo deste artigo.

Parâmetros com valores de tabela (TVPs)

O mssql-python driver não suporta parâmetros com valores de tabela. Para a matriz completa de suporte a recursos, veja ciclo de vida de suporte. Se você precisar passar um conjunto de linhas para um procedimento armazenado, use estas alternativas:

Insira primeiro em uma tabela temporária e, em seguida, faça o procedimento armazenado lê-la.
Use bulkcopy() para carregar dados em uma tabela de preparação.
Passe uma string JSON e faça o parse dela com OPENJSON dentro do procedimento.

Comentários

Esta página foi útil?

Última atualização em 2026-11-05