使用pytds在Python中调用存储过程插入SQL Server表时遇TypeError求助

错误信息 #

Traceback (most recent call last):
File "c:/Users/mydesk/Desktop/test/test_pytds.py", line 64, in 
connect_and_call_stored_procedure(query, val)
File "c:/Users/mydesk/Desktop/test/test_pytds.py", line 48, in call_sproc
cursor.execute(query, (tvp,))
File "C:\Program Files (x86)\Python38-32\lib\site-packages\pytds\__init__.py", line 739, in execute
self._execute(operation, params)
File "C:\Program Files (x86)\Python38-32\lib\site-packages\pytds\__init__.py", line 701, in _execute
operation = operation % names[0]
TypeError: not enough arguments for format string

问题代码 #

import pytds 
def get_query() -> str:
    sql = '''
        DECLARE @tbl [dbo].[tbl]
        insert into @tbl
        (
            col1, col2, col3
        )
        values (%s,%s,%s)
    '''  
    query += '''
         EXECUTE [dbo].[stproc] 
         @tbl
    '''
    return sql

def call_sproc(query, val):
    server = 'my_server'
    user = 'my_user'
    password = 'secret'
    database = 'my_db'
    
    try:
        conn = pytds.connect(server=server, user=user, password=password, database=database)
        cursor = conn.cursor()

        tvp = pytds.TableValuedParam(type_name='dbo.tbl', rows=val)

        cursor.execute(query, (tvp,))

        conn.commit()

        cursor.close()
        conn.close()

    except pytds.Error as e:
        print("Connection issue:", e)

###
val=[('col1_val'), ('col2_val'), ('col3_val')]
my_sql = get_query()
call_sproc(my_sql, val)

问题分析与解决 #

• 函数变量未定义：get_query里初始化了sql变量，但后续用query +=拼接，query未定义会触发NameError，需改为sql +=后返回sql。
• 占位符与参数不匹配：SQL语句里有3个%s占位符，但执行时仅传入1个TVP参数，这是触发TypeError的直接原因。实际无需手动在SQL中INSERT表变量，表值参数（TVP）可直接传递给存储过程。
• TVP使用逻辑错误：pytds的TableValuedParam是直接作为参数传给存储过程的，不需要先DECLARE表变量再插入数据。存储过程[dbo].[stproc]应定义了接收dbo.tbl类型的参数，直接传TVP即可。
• 数据格式错误：val中的('col1_val')是字符串而非元组，正确的单条数据格式应为('col1_val', 'col2_val', 'col3_val')，多条数据则是[('v1','v2','v3'), ('v4','v5','v6')]。

修正后的代码示例 #

import pytds 

def get_query() -> str:
    # 直接调用存储过程，接收表值参数
    return '''
        EXECUTE [dbo].[stproc] @tbl = %s
    '''

def call_sproc(query, val):
    server = 'my_server'
    user = 'my_user'
    password = 'secret'
    database = 'my_db'
    
    cursor = None
    conn = None
    try:
        conn = pytds.connect(server=server, user=user, password=password, database=database)
        cursor = conn.cursor()

        # 传入符合要求的表值参数
        tvp = pytds.TableValuedParam(type_name='dbo.tbl', rows=val)
        cursor.execute(query, (tvp,))
        conn.commit()

    except pytds.Error as e:
        print("Connection issue:", e)
    finally:
        # 确保资源被正确释放
        if cursor:
            cursor.close()
        if conn:
            conn.close()

# 单条数据示例
val=[('col1_val', 'col2_val', 'col3_val')]
# 多条数据示例：
# val=[('v1','v2','v3'), ('v4','v5','v6')]
my_sql = get_query()
call_sproc(my_sql, val)

内容的提问来源于stack exchange，提问作者RSG