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

Question

У меня есть хранимая процедура на сервере MSSQL, "prc_add_names", которая принимает параметр табличного значения. Сам параметр имеет пользовательский тип "StringTable", определенный следующим образом:

CREATE TYPE [dbo].[StringTable] AS TABLE([strValue] [nvarchar](max) NULL)

Я понятия не имею, как выполнить эту процедуру, используя SQLAlchemy. Я привык вызывать процедуры с использованием аргументов session.execute как это:

result = session.execute('prc_do_something :pArg', {pArg:'foo'})

Однако это не сработает, если я просто передам список строк в качестве аргумента:

result = session.execute('prc_add_names :pArg', {pArg: ['Name One', 'Name Two']})

что приводит к:

sqlalchemy.exc.ProgrammingError: (pymssql.ProgrammingError) (102, "Incorrect syntax near 'Name One'.DB-Lib error message 20018, severity 15:
General SQL Server error: Check messages from the SQL Server
") [SQL: 'prc_add_names %(pArg)s'] [parameters: {'pArg': ['Name One', 'Name Two']}] (Background on this error at: http://sqlalche.me/e/f405)

Очевидно, что SQLAlchemy не воспринимает мой список строк как попытку создать мой аргумент типа StringTable, но после пары часов поиска в Google и чтения документации я так и не понял, как мне поступить с этим.

К вашему сведению, я не контролирую эту базу данных, поэтому изменение хранимой процедуры или чего-то еще там не вариант.

РЕДАКТИРОВАТЬ: я не женат на SQLAlchemy. Если есть другая библиотека, которая может справиться с этим, я был бы рад использовать ее вместо этого.

Answer 1

Есть драйвер, который действительно поддерживает TVP: Pytds. Официально это не поддерживается, но есть сторонняя реализация диалекта: sqlalchemy-pytds. Используя их, вы можете вызвать хранимую процедуру следующим образом:

In [1]: engine.execute(DDL("CREATE TYPE [dbo].[StringTable] AS TABLE([strValue] [nvarchar](max) NULL)"))
Out[1]: <sqlalchemy.engine.result.ResultProxy at 0x7f235809ae48>

In [2]: engine.execute(DDL("CREATE PROC test_proc (@pArg [StringTable] READONLY) AS BEGIN SELECT * FROM @pArg END"))
Out[2]: <sqlalchemy.engine.result.ResultProxy at 0x7f2358027b70>

In [3]: arg = ['Name One', 'Name Two']

In [4]: import pytds

In [5]: tvp = pytds.TableValuedParam(type_name='StringTable',
   ...:                              rows=((x,) for x in arg))

In [6]: engine.execute('EXEC test_proc %s', (tvp,))
Out[6]: <sqlalchemy.engine.result.ResultProxy at 0x7f294e699e10>

In [7]: _.fetchall()
Out[7]: [('Name One',), ('Name Two',)]

Таким образом, вы можете передавать потенциально большие объемы данных в виде параметров:

In [21]: tvp = pytds.TableValuedParam(type_name='StringTable',
    ...:                              rows=((str(x),) for x in range(100000)))

In [22]: engine.execute('EXEC test_proc %s', (tvp,))
Out[22]: <sqlalchemy.engine.result.ResultProxy at 0x7f294c6e9f98>

In [23]: _.fetchall()[-1]
Out[23]: ('99999',)

Если, с другой стороны, вы используете драйвер, который не поддерживает TVP, вы можете объявить переменную таблицы, вставить значения и передать ее в качестве аргумента вашей процедуре:

In [12]: engine.execute(
    ...:     """
    ...:     DECLARE @pArg AS [StringTable];
    ...:     INSERT INTO @pArg VALUES {placeholders};
    ...:     EXEC test_proc @pArg;
    ...:     """.format(placeholders=",".join(["(%s)"] * len(arg))),
    ...:     tuple(arg))
    ...:     
Out[12]: <sqlalchemy.engine.result.ResultProxy at 0x7f23580f2908>

In [15]: _.fetchall()
Out[15]: [('Name One',), ('Name Two',)]

Обратите внимание, что вы не можете использовать любые методы executemany, иначе вы в конечном итоге вызовете процедуру для каждого табличного значения отдельно. Вот почему заполнители создаются вручную, а значения таблиц передаются как отдельные аргументы. Следует позаботиться не о том, чтобы форматировать какие-либо аргументы непосредственно в запросе, а о правильном количестве заполнителей для DB-API. Значения строк ограничены максимумом 1000.

Конечно, было бы неплохо, если бы основной драйвер DB-API обеспечивал надлежащую поддержку табличных параметров, но по крайней мере я не смог найти способ для pymssql, который использует FreeTDS. Ссылка на TVP в списке рассылки дает понять, что они не поддерживаются. Ситуация не намного лучше для PyODBC.

_{Отказ от ответственности: я действительно не использовал MS SQL Server раньше.}

Answer 2

Я думаю, что вы можете использовать callproc() метод и передать входной аргумент в виде списка. http://docs.sqlalchemy.org/en/latest/core/connections.html