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

Executar um procedimento armazenado utilizando a sequência de escape {CALL}. Este exemplo chama o sp_databases procedimento armazenado do sistema, que não aceita parâmetros e devolve uma linha por base 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 marcadores 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 devolver valores através de parâmetros de saída. Use Transact-SQL variáveis (T-SQL) para capturar valores de saída e depois recupere-as com uma SELECT instrução:

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

Valores de retorno

Capturar o valor de retorno do procedimento armazenado

Executar um procedimento armazenado que devolve 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 devolve 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 avançar, nextset() atualiza rowcount. Um resultado de modificação de dados reporta a sua própria contagem de linhas afetadas e tem description is None; não chame métodos fetch para esse resultado. Um resultado que produz uma linha começa com rowcount == -1 antes de buscar. SET NOCOUNT ON suprime os resultados de linhas afetadas. Para um lote que mistura estes tipos de resultados, veja Acompanhar cada resultado num lote.

Alguns procedimentos armazenados retornam múltiplos 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 devolvidos 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

Controlo explícito de transações

Agrupe múltiplas chamadas de procedimento armazenado numa 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 gerir a transação

Se o procedimento armazenado gerir as 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 em procedimentos armazenados

Tratar 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 comandos PRINT e mensagens informativas

As instruções SQL Server PRINT e RAISERROR com gravidade inferior a 11 são registadas em cursor.messages após a execução. Cada entrada é um (message_type, message_text) tuplo. Quando um PRINT é executado antes de um conjunto de resultados, ocupa um conjunto de resultados sem linhas próprio, por isso leia primeiro cursor.messages e, em seguida, chame nextset() para aceder às 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 através de múltiplos conjuntos de resultados, leia cursor.messages após execute() e novamente após cada nextset() chamada para que as 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 Gestão de cursores.

Melhores práticas

Usar parâmetros nomeados

Os parâmetros nomeados são mais claros e preservam a independência 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"))

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

Utilizar SET NOCOUNT ON em procedimentos armazenados

Para um melhor desempenho e uma gestão de resultados mais limpa, certifique-se de que os seus procedimentos armazenados incluem:

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 obter um valor de identidade de um INSERT (com ou sem procedimento guardado), utilize OUTPUT INSERTED em vez de SCOPE_IDENTITY(). Esta abordagem devolve 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}")

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

Funcionalidades não suportadas

callproc()

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

Parâmetros com valor de tabela (TVPs)

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

Insere primeiro numa tabela temporária e depois faz com que o procedimento armazenado seja lido dela.
Use bulkcopy() para carregar dados numa tabela de staging.
Passe uma cadeia JSON e faça o respetivo parse com OPENJSON dentro do procedimento.

Comentários

Esta página foi útil?

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