Python call sql-server stored procedure with table valued parameter

Question

I have a python script that loads , transform and calculates data. In sql-server there's a stored procedure that requires a table valued parameter, 2 required parameters and 2 optional parameters. In sql server I can call this 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

after a comprehensive search I found the following post:

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

it suggests to use PYTDS and the sql-alchemy 's dialect 'sql alchemy pytds' to call a SP with table valued parameters. with this post and the documentation I created the following Python script:

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', 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="MergeOnColumn", rows=(arg))
cur.execute('EXEC test_proc %s', ("[dbname].[table2]", "[dbname].[table1]", tvp,))
cur.fetchall()

When I run this code I get the following error message:

TypeError: not all arguments converted during string formatting

Doe anyone know how to pass in the multiple arguments correctly or has a suggestion how I could handle this call SP directly?

Answer 1

1. Below is the python code to pass values into table values parameter of stored procedure and get output parameter and print the return message
2. First need to create user defined table table type.
3. Need to declare a table valued input parameter as well in Stored procedure.
4. Same table type needs to use in the sql query using in the python code to execute the stored procedure.

import pandas as pd, pyodbc, string
rd = pd.read_csv(D:\ColumnList.csv)

conn = pyodbc.connect(Driver='{SQL Server}',Server="your Server Name", Database="DB Name", Trusted_Connection='yes', autocommit=True)
                            
cursor = conn.cursor()

values = [tuple(row) for row in rd.iloc[1:].values]
    
#remove '[' and ']' from tuple so that it can be in format of insert query
values = str(values).replace('[','').replace(']','')

cursor.execute("create table #SPReturnVal(ReturnValue varchar(500))")
 
query = "declare @table as tabletype insert into @table values"+values+"declare  @ReturnMessage varchar(500)"+" exec  USP_GetRecordsCount @TableListFromSheet=@table, @ReturnMessage=@ReturnMessage out insert into #SPReturnVal values(@ReturnMessage)"+"" 

cursor.execute("select ReturnValue from #SPReturnVal")

#fetch value from cursor
message=cursor.fetchval()

print(message)

Answer 2

mssql+pymssql://

pymssql does not directly support TVPs (issue here), but for SQL Server 2016+ there is a workaround.

Given a user-defined table type

CREATE TYPE dbo.tvp_table AS TABLE 
(
    id int NOT NULL, 
    txt nvarchar(10) NOT NULL, 
    PRIMARY KEY (id)
)

and a stored procedure that consumes it

CREATE PROCEDURE dbo.tvp_echo 
    @tvp_in dbo.tvp_table READONLY
AS
BEGIN
    SET NOCOUNT ON;
    SELECT * FROM @tvp_in;
END

we can do:

tvp_columns = ["id", "txt"]
tvp_data = [(1, "foo"), (2, "bar")]
tvp_json = [dict(zip(tvp_columns, row)) for row in tvp_data]
print(tvp_json)
# [{'id': 1, 'txt': 'foo'}, {'id': 2, 'txt': 'bar'}]

sql = """\
SET NOCOUNT ON;
DECLARE @tvp dbo.tvp_table;

INSERT INTO @tvp
SELECT id, txt
FROM OPENJSON(%s)
WITH (
    id int '$.id',
    txt nvarchar(10) '$.txt'
);

EXEC dbo.tvp_echo @tvp
"""
crsr.execute(sql, (json.dumps(tvp_json, default=str),))
print(crsr.fetchall())
# [(1, 'foo'), (2, 'bar')]

Answer 3

mssql+pyodbc://

pyodbc added support for table-valued parameters (TVPs) in version 4.0.25, released 2018-12-13. Simply supply the TVP value as a list of tuples:

proc_name = "so51930062"
type_name = proc_name + "Type"

# set up test environment
with engine.begin() as conn:
    conn.exec_driver_sql(f"""\
        DROP PROCEDURE IF EXISTS {proc_name} 
    """)
    conn.exec_driver_sql(f"""\
        DROP TYPE IF EXISTS {type_name} 
    """)
    conn.exec_driver_sql(f"""\
        CREATE TYPE {type_name} AS TABLE (
        id int,
        txt nvarchar(50)
        ) 
    """)
    conn.exec_driver_sql(f"""\
        CREATE PROCEDURE {proc_name} 
        @prefix nvarchar(10),
        @tvp {type_name} READONLY
        AS
        BEGIN
            SET NOCOUNT ON;
            SELECT id, @prefix + txt AS new_txt FROM @tvp;
        END
    """)

#run test
with engine.begin() as conn:
    data = {"prefix": "new_", "tvp": [(1, "foo"), (2, "bar")]}
    sql = f"{{CALL {proc_name} (:prefix, :tvp)}}"
    print(conn.execute(sa.text(sql), data).fetchall())
    # [(1, 'new_foo'), (2, 'new_bar')]

Answer 4

pytds

Python DBAPI driver for MSSQL using pure Python TDS (Tabular Data Stream) protocol implementation

I used pytds for merge / upsert via a stored procedure targeting a SQL Server.

Example

Here are a example of the basic functions, a row data is represented by Tuple:

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)

Answer 5

On the basis of the comments to my question i've managed to get the stored procedure running with table valued parameters (and get the return values from the SP) The final script is as follows:

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)

The autocommit is added in the connection (to commit the transaction in the cursor), the table valued parameter (marchcolumntable) expects 2 columns, so the arg is modified to fit 2 columns.

The parameters that are required besides the tvp are included in the exec string. The last param in the execute string is the name of the tvp parameter(mergeoncolumn) that is filled with the tvp.

optionally you can add the result status or row count as descripted in the pytds documentation: https://python-tds.readthedocs.io/en/latest/index.html

Note!: in the stored procedure you have to make sure that the SET NOCOUNT ON is added otherwise you wont get any results back to Python