mssql-python으로 저장된 프로시저를 호출하기

mssql-python 드라이버는 DB-API 2.0 명세의 메서드를 구현 callproc() 하지 않습니다. callproc()를 호출하면 NotSupportedError가 발생합니다. 대신, 표준 쿼리 실행 메서드와 함께 ODBC {CALL ...} 이스케이프 시퀀스를 사용하세요.

기본 저장 프로시저 실행

매개변수 없이

이스케이프 시퀀스를 사용하여 {CALL} 저장 프로시저를 실행합니다. 이 예시는 매개변수를 받지 않고 데이터베이스당 한 행씩 반환하는 시스템 저장 프로시저를 sp_databases 호출합니다:

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)

입력 매개변수 포함

위치별 또는 명명된 자리 표시자를 사용하여 매개변수를 전달합니다:

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

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

출력 매개 변수

출력 매개변수를 선언하고 가져오기

SQL Server 저장 프로시저는 출력 매개변수를 통해 값을 반환할 수 있습니다. Transact-SQL(T-SQL) 변수를 사용해 출력 값을 캡처한 후, 다음과 같은 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}")

다중 출력 매개변수

개별 변수를 선언하여 여러 출력 값을 캡처합니다. 동일한 패턴은 매개변수가 있는 OUTPUT 모든 저장 프로시저에도 적용됩니다:

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

반환 값

저장 프로시저의 반환 값 캡처

결과를 반환하는 저장 프로시저를 실행합니다:

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

출력 매개변수가 포함된 반환 값

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

결과 집합

단일 결과 집합

단일 결과 집합을 반환하는 저장 프로시저를 실행합니다:

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

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

다중 결과 집합

일부 저장 프로시저는 여러 결과 집합을 반환합니다. 둘 사이를 이동하려면 nextset()를 사용하세요:

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

더 많은 결과 세트를 확인해 보세요

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

저장 프로시저가 포함된 트랜잭션

명시적 거래 제어

원자성을 보장하기 위해 여러 저장 프로시저 호출을 트랜잭션으로 묶으세요:

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

저장 프로시저가 트랜잭션을 관리하도록 하라

저장 프로시저가 자체 트랜잭션을 처리한다면:

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

오류 처리

저장 프로시저 오류 포착

저장 프로시저나 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}")

PRINT 문과 정보 메시지 캡처

SQL Server PRINT 구문과 심각도가 11 미만인 RAISERROR은 실행 후 cursor.messages에 캡처됩니다. 각 항목은 (message_type, message_text) 튜플입니다. PRINT가 결과 집합보다 먼저 실행되면 자체적으로 행이 없는 결과 집합 하나를 차지하므로, 먼저 cursor.messages를 읽은 다음 nextset()를 호출하여 행에 도달합니다:

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

저장 프로시저가 여러 결과 집합에 걸쳐 메시지를 내보낼 PRINT 때, 모든 결과 집합의 메시지가 수집되도록 execute() 후에 cursor.messages를 읽고, 각 nextset() 호출 후에도 다시 읽습니다:

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

전체 cursor.messages API는 커서 관리 항목을 참조하세요.

모범 사례

이름 있는 매개변수 사용

명명된 매개변수는 더 명확하며 순서에 독립적입니다:

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

nullable 출력 매개변수 처리

저장 프로시저에서 출력 매개변수를 가져올 때 NULL 값을 확인하세요:

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

저장 프로시저에서 ON의 사용 SET NOCOUNT

더 나은 성능과 깔끔한 결과 처리를 위해 저장 프로시저에 다음을 포함하도록 하세요:

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

예시: 완전한 워크플로우

다음은 저장 프로시저를 호출하고, 출력 값을 가져오며, 오류를 처리하는 실용적인 예시입니다:

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)

OUTPUT INSERTED를 사용하여 생성된 키를 얻습니다

저장 프로시저 사용 여부와 관계없이 INSERT에서 ID 값을 검색하려면 SCOPE_IDENTITY() 대신 OUTPUT INSERTED을 사용하세요. 이 방법은 동일한 결과 집합에서 값을 반환합니다:

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

이 패턴은 정체성 열이 있는 모든 테이블에 적용되며 저장 프로시저가 필요하지 않습니다.

지원되지 않는 기능

callproc()

cursor.callproc()를 호출하면 mssql-python 드라이버가 NotSupportedError를 발생시킵니다. 이 문서 전체에서 볼 수 있듯이 대신 cursor.execute("{CALL ...}") 또는 cursor.execute("EXECUTE ...")을 사용하세요.

테이블 값 매개변수(TVP)

테이블 값 매개변수는 현재 버전(1.11.0) mssql-python에서 지원되지 않습니다. 저장 프로시저에 행 집합을 전달해야 한다면, 다음과 같은 대안을 사용하세요:

먼저 임시 테이블에 삽입한 후, 저장된 프로시저를 읽게 하세요.
스테이징 테이블에 데이터를 로드하는 데 사용 bulkcopy() 하세요.
JSON 문자열을 전달하고 프로시저 내에서 OPENJSON를 사용하여 이를 파싱합니다.

피드백

이 페이지가 도움이 되었나요?

Last updated on 2026-07-22