Cómo llamar al procedimiento almacenado con SQLAlchemy que requiere un parámetro de tabla de tipo definido por el usuario

Tengo un procedimiento almacenado en el servidor MSSQL, “prc_add_names”, que toma un parámetro de valor de tabla. El parámetro en sí es de un tipo personalizado “StringTable” definido así:

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

No tengo idea de cómo ejecutar este procedimiento utilizando SQLAlchemy. Estoy acostumbrado a llamar a procedimientos con argumentos usando session.execute esta manera:

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

Sin embargo, esto no funciona si simplemente paso una lista de cadenas como argumento:

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

lo que lleva a:

 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) 

Obviamente, SQLAlchemy no entiende mi lista de cadenas como un bash de crear mi argumento de tipo StringTable, pero después de un par de horas de buscar en Google y leer la documentación, no he descubierto cómo debo manejar esto.

Para su información, no tengo el control de esta base de datos, por lo que la modificación del procedimiento almacenado o cualquier otra cosa no es una opción.

EDIT : no estoy casado con SQLAlchemy. Si hay otra biblioteca que puede manejar esto, me encantaría usarla en su lugar.

Hay un controlador que realmente soporta TVPs: Pytds . No está soportado oficialmente, pero hay una implementación de dialecto de terceros: sqlalchemy-pytds . Usándolos puedes llamar a tu procedimiento almacenado así:

 In [1]: engine.execute(DDL("CREATE TYPE [dbo].[StringTable] AS TABLE([strValue] [nvarchar](max) NULL)")) Out[1]:  In [2]: engine.execute(DDL("CREATE PROC test_proc (@pArg [StringTable] READONLY) AS BEGIN SELECT * FROM @pArg END")) Out[2]:  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]:  In [7]: _.fetchall() Out[7]: [('Name One',), ('Name Two',)] 

De esta manera puede pasar cantidades potencialmente grandes de datos como parámetros:

 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]:  In [23]: _.fetchall()[-1] Out[23]: ('99999',) 

Si, por otro lado, está utilizando un controlador que no es compatible con TVP, podría declarar una variable de tabla , insertar los valores y pasarlo como argumento a su procedimiento:

 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]:  In [15]: _.fetchall() Out[15]: [('Name One',), ('Name Two',)] 

Tenga en cuenta que no puede usar ningún método de ejecución, o terminará llamando al procedimiento para cada valor de tabla por separado. Es por eso que los marcadores de posición se construyen manualmente y los valores de la tabla se pasan como argumentos individuales. Se debe tener cuidado de no formatear ningún argumento directamente en la consulta, sino la cantidad correcta de marcadores de posición para la DB-API. Los valores de fila están limitados a un máximo de 1000 .

Por supuesto, sería bueno si el controlador DB-API subyacente proporcionara el soporte adecuado para los parámetros con valores de tabla, pero al menos no pude encontrar una forma para pymssql, que usa FreeTDS. Una referencia a los TVP en la lista de correo deja claro que no son compatibles. La situación no es mucho mejor para PyODBC .

Descargo de responsabilidad: no he usado MS SQL Server antes.

Creo que puedes usar el método callproc() y pasar el argumento de entrada como una lista. http://docs.sqlalchemy.org/en/latest/core/connections.html#calling-stored-procedures