Python使用表值param调用sqlserver存储过程

Question

我有一个python脚本来加载、转换和计算数据。在SQLServer中，有一个存储过程需要一个表值参数、两个必需参数和两个可选参数。在sql server中，我可以将此SP称为：

USE [InstName]
GO

DECLARE @return_value int
DECLARE @MergeOnColumn core.MatchColumnTable

INSERT INTO @MergeOnColumn
SELECT 'foo.ExternalInput','bar.ExternalInput'

EXEC    @return_value = [core].[_TableData]
        @Target = N'[dbname].[tablename1]',
        @Source = N'[dbname].[table2]',
        @MergeOnColumn  = @MergeOnColumn,
        @Opt1Param = False,
        @Opt2Param = False

SELECT  'Return Value' = @return_value

GO

经过全面搜索，我发现了以下帖子：

How to call stored procedure with SQLAlchemy that requires a user-defined-type Table parameter

它建议使用PYTDS和sql alchemy的方言“sql alchemy PYTDS”来调用带有表值参数的SP。 通过这篇文章和文档，我创建了以下Python脚本：

^{pr2}$

运行此代码时，我收到以下错误消息：

TypeError: not all arguments converted during string formatting

有人知道如何正确地传递多个参数吗？或者有人建议我如何直接处理这个调用SP？在

Answer 1

根据对我的问题的评论，我设法让存储过程使用表值参数运行（并从SP获取返回值） 最后的脚本如下：

import pandas as pd
import pytds
from pytds import login
import sqlalchemy as sa
from sqlalchemy import create_engine
import sqlalchemy_pytds

def connect():
    return pytds.connect(dsn='ServerName',database='DBName',autocommit=True, auth=login.SspiAuth())

engine = sa.create_engine('mssql+pytds://[ServerName]', creator=connect)
conn = engine.raw_connection()

with conn.cursor() as cur:
    arg = [["foo.ExternalInput","bar.ExternalInput"]]
    tvp = pytds.TableValuedParam(type_name="core.MatchColumnTable", rows=arg)
    cur.execute("EXEC test_proc @Target = N'[dbname].[tablename1]', @Source = N'[dbname].[table2]', @CleanTarget = 0, @UseColumnsFromTarget = 0, @MergeOnColumn = %s", (tvp,))
    result = cur.fetchall()
    print(result)

autocommit被添加到连接中（在游标中提交事务），表值参数（marchcolumntable）需要2列，因此修改arg以适合2列。在

除了tvp之外还需要的参数包含在exec字符串中。execute字符串中的最后一个参数是用tvp填充的tvp参数（mergeoncolumn）的名称。在

您可以选择添加pytds文档中描述的结果状态或行数： https://python-tds.readthedocs.io/en/latest/index.html

注意！：在存储过程中，必须确保 将添加setnocount ON，否则将无法将任何结果返回到Python

Answer 2

pytds公司

使用纯Python TDS（表格数据流）协议实现的MSSQL的Python DBAPI驱动程序

我使用pytds作为merge / upsert，通过一个以SQL服务器为目标的stored procedure。在

示例

以下是基本函数的示例，行数据由元组表示：

def get_connection(instance: str, database: str, user: str, password: str):
    return pytds.connect(
        dsn=instance, database=database, user=user, password=password, autocommit=True
    )

def execute_with_tvp(connection: pytds.Connection, procedure_name: str, rows: list):
    with connection.cursor() as cursor:
         tvp = pytds.TableValuedParam(type_name=my_type, rows=rows)
         cursor.callproc(procedure_name, tvp)

仅供参考

对于我的最后一个解决方案，我离开了python并实现了一个c程序，因为python的性能对于一个体面的ETL来说是很低的，而且pytds不支持使用非标准端口连接到命名实例。Checkout the issue I created for updates.