Python-вызов sql-server хранимой процедуры с табличным параметром

Question

У меня есть скрипт Python, который загружает, преобразовывает и рассчитывает данные. В sql-сервере есть хранимая процедура, для которой требуется табличный параметр, 2 обязательных параметра и 2 необязательных параметра. В 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

после всестороннего поиска я нашел следующий пост:

Как вызвать хранимую процедуру с помощью SQLAlchemy, для которой требуется параметр таблицы пользовательского типа

он предлагает использовать PYTDS и диалект sql-alchemy 'sql alchemy pytds' для вызова SP с табличными параметрами. с помощью этого поста и документации я создал следующий скрипт Python:

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

Когда я запускаю этот код, я получаю следующее сообщение об ошибке:

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)

Автокоммит добавляется в соединение (для фиксации транзакции в курсоре), табличный параметр (marchcolumntable) ожидает 2 столбца, поэтому аргумент arg изменяется так, чтобы он соответствовал 2 столбцам.

Параметры, которые требуются помимо tvp, включены в строку exec. Последний параметр в строке выполнения - это имя параметра tvp (mergeoncolumn), который заполняется значением tvp.

При желании вы можете добавить статус результата или количество строк, как описано в документации по pytds: https://python-tds.readthedocs.io/en/latest/index.html

Заметка!: в хранимой процедуре вы должны убедиться, что SET NOCOUNT ON добавлен, иначе вы не получите никаких результатов обратно в Python

Answer 2

pytds

Драйвер Python DBAPI для MSSQL с использованием чистой реализации протокола Python TDS (Tabular Data Stream)

Я использовал pytds для слияния / обновления через хранимую процедуру, ориентированную на SQL Server.

пример

Вот пример основных функций, данные строки представлены 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)

К вашему сведению

Для своего окончательного решения я отошел от python и реализовал программу на C #, потому что производительность python была слишком низкой для достойного ETL, а также из-за того, что швы pytds не поддерживали подключение к именованному экземпляру с использованием нестандартного порта. Проверьте созданную мной проблему для обновлений.

Answer 3

mssql+pyodbc://

pyodbc добавил поддержку параметров с табличным значением (TVP) в версии 4.0.25, выпущенной 13 декабря 2018 г. Просто укажите значение TVP в виде списка кортежей:

      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

mssql+pymssql://

pymssql не поддерживает TVP напрямую (проблема здесь ), но для SQL Server 2016+ есть обходной путь.

Учитывая пользовательский тип таблицы

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

и хранимая процедура, которая ее использует

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

Мы сможем:

      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 5

1. Ниже приведен код Python для передачи значений в параметр табличных значений хранимой процедуры, получения выходного параметра и печати возвращаемого сообщения.
2. Сначала необходимо создать таблицу пользовательского типа.
3. Необходимо также объявить входной параметр с табличным значением в хранимой процедуре.
4. Тот же тип таблицы необходимо использовать в запросе sql, используя код Python для выполнения хранимой процедуры.

      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)