mssql-pythonでストアドプロシージャを呼び出す

mssql-pythonドライバーは DB-API 2.0仕様の callproc() メソッドを実装していません。 callproc()に電話するとNotSupportedErrorが上がる。代わりに、標準的なクエリ実行メソッドでODBC {CALL ...} エスケープシーケンスを使用します。

基本的なストアドプロシージャ実行

パラメータなし

{CALL}エスケープシーケンスを使ってストアドプロシージャを実行します。この例は sp_databases システムのストアドプロシージャを呼び出し、パラメータを取らず、データベースごとに1行を返します。

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メッセージを送信した場合、cursor.messagesの後と各execute()呼び出しの後に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")

ストアドプロシージャで SET NOCOUNT ONを活用してください

より良いパフォーマンスとクリーンな結果処理のために、ストアドプロシージャに以下を含めるようにしてください:

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から識別値を取得するには(ストアドプロシージャの有無にかかわらず)、OUTPUT INSERTEDの代わりにSCOPE_IDENTITY()を使います。この方法は同じ結果セットで値を返します:

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