Hello Community!
I have not posted here for several years as I have been getting along just
fine with the excellent SQLAlchemy toolkit and the excellent documentation.
However, I am trying to do something a bit fancy since all the SQL Server
dataservers I use are running versions that support OPENJSON. A number of
our applications have used staging techniques to copy tables of data to the
dataserver so that that can be used in JOINs (e.g. send up 10,000
identifiers into the table rather than doing RBAR or batching into IN
clauses). This technique is IO intensive and has quite a bit of overhead.
An alternative approach is to serialize the data into a big blob of JSON,
send it over as a single scalar and then expand it out to a table via a CTE
on the dataserver. We have found the performance of this to be excellent
from scripting language, presumably because we are avoiding so many API
calls. However, I have not found a way to do this cleanly from SQLAlchemy
and end up using sql.text a lot. Also, because it does not seem possible to
re-use the same placeholder multiple times in a statement, if we want to
expand out the JSON multiple times (e.g. if the JSON is nested and has
multiple tables in it), we work around that using [sys].[sp_executesql].
Please see snippet below and a complete script (with create_engine
redacted) attached.
What I am looking for is some guidance on how to use SQLALchemy to produce
CTEs that reference the same variable (@blob_o_json) multiple times without
having to pass in the same value multiple times.
thanks in advance!
pjjH
# TODO
: formulate this as a SQLAlchemy expression (is there an exec() element?)
outer_sql = r'''
EXEC [sys].[sp_executesql] @statement = :sql,
@params = N'@bind_value_blob_o_json varchar(max)',
@bind_value_blob_o_json = :json_blob
'''
# we prepare the *outer* statement that has a single command in it
statement = sql.text(outer_sql)
# now execute that passing in unicode value of the inner SQL.
logging.basicConfig(format='%(asctime)s %(message)s')
logging.getLogger().setLevel(logging.DEBUG)
nrows = [1000,10000,100000, 200000, 500000]
for n in nrows:
# we cons up a list of dicts to represent a single 'inline' table with lots of
rows
logging.debug("passing %d items" % (n))
blob_o_json =json.dumps([dict(rn=x,foo=x*2,bar=x-1,fruit='banana',
flavor='yummy') for x in
range(0,n)])
# Note: we named arguments in this next line correspond to the :sql and
:json_blob placeholders in outer_sql
result = engine.execute(statement,sql=six.text_type(inner_sql),
json_blob=blob_o_json)
rs = result.fetchall()
#logging.debug((len(rs), rs[]))
# drop down to DBAPI to process each result-set. The application should know
how many result-sets will be present
# I think it is likely that the vast majority of queries will just have a
single result-set
# I did not know how to get to anything other than the first result-set ... I
thought there
# was probably some pyodbc magic to make it work with SQLAlchemy and -- sure
enough! -- there
# was a ticket with a workaround:
https://github.com/sqlalchemy/sqlalchemy/issues/1635#issuecomment-441907416
#cursor = result.cursor
#rs1 = cursor.fetchall()
#cursor.nextset()
#rs2 = cursor.fetchall()
#logging.debug((len(rs1), rs1[-4:], rs2[-4:]))
logging.debug("finished %d" % (n))
logging.debug("done")
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To view this discussion on the web visit
https://groups.google.com/d/msgid/sqlalchemy/345f67ce-a4b9-4575-87bc-81653868908en%40googlegroups.com.
import six
import json
import logging
import urllib
import sqlalchemy
from sqlalchemy import create_engine, sql
# we are using the SQL Express 2017 'metatweaker' instance as that is a readily accessible
# engine that is running a recent enough revision of SQL Server ... OPENJSON is not in SQL 2014
# Also, if you want to see what the performance characteristics of this look like on the server side, you need to
# have some roles such as VIEW SERVER STATE (e.g. to look at query plans)
# engine = create_engine("mssql+pyodbc://@dbreins.deshaw.com\DBREINS/master?driver=SQL+Server")
# engine = create_engine(r'mssql+pyodbc://@fretsaw.deshaw.com\sqlexpress/metatweaker?driver=SQL+Server')
#engine = create_engine(r"mssql+pyodbc://@frosthawk.deshaw.com\sqlexpress2017/metatweaker?driver=SQL+Server")
#engine = create_engine(r'mssql+pyodbc://@dbreinsdev.deshaw.com\reinsdev,1440/reins?APP=sqlalchemy')
engine = create_engine(r'mssql+pyodbc://@dbreinsdev.deshaw.com\reinsdev:1440/reins?driver=ODBC+Driver+11+for+SQL+Server&APP=sqlalchemy')
#quoted=urllib.parse.quote_plus('Driver={Easysoft ODBC-SQL Server};Server=dbreinshyd3.deshaw.com;port=1433;Application=Python;Kerberos=1;ServerSPN=MSSQLSvc/dbreinshyd3.deshaw.com:1433')
#engine = create_engine('mssql+pyodbc:///?odbc_connect={}'.format(quoted))
inner_sql = r'''
-- do we want to allow multiple blobs of JSON?
declare @blob_o_json varchar(max);
select @blob_o_json=@bind_value_blob_o_json; -- Hopefully this is not string-substituted but, rather, a prepared statement is created
-- with one bind value per entry in @params
-- yay! we can put comments in here too
-- TODO: figure out how to get SQLAlchemy to combine the variable declaration and the CTEs.
WITH T(foo,give_n_a_funny_alias) as (
SELECT foo,give_n_a_funny_alias
FROM OPENJSON(@blob_o_json) WITH (
give_n_a_funny_alias int 'strict $.rn',
foo int 'strict $.foo'
) AS alias_for_the_selectable_from_OPENJSON
),T1(n,foo) as (
SELECT n,foo
FROM OPENJSON(@blob_o_json) WITH (
n int 'strict $.rn',
foo int 'lax $.no_I_am_not_here_in_the_JSON' -- this key is not present ... is we use strict rather than lax, the query would blow up
) AS alias_for_the_selectable_from_OPENJSON
)
SELECT T.foo, T.give_n_a_funny_alias as bar
FROM T -- A CTE derived from @blob_o_json
JOIN dbo.Nums -- JOINing on a vanilla table in the database
ON (T.give_n_a_funny_alias=Nums.n)
JOIN T1 -- Another CTE from the JSON
ON (T1.n = Nums.n)
'''
# Note that the named arguments here match up with the :param_name placeholders in the *outer*
# statement. The @params value says that the parameters of the *inner* SQL are 'bind_value_blob_json'.
# This expansion of parameters at different stages is a little difficult to follow but one should not have to write this kind of code multiple times.
# It looks like variable declarations work in the inner SQL and each statement is executed.
# TODO: formulate this as a SQLAlchemy expression (is there an exec() element?)
outer_sql = r'''
EXEC [sys].[sp_executesql] @statement = :sql,
@params = N'@bind_value_blob_o_json varchar(max)',
@bind_value_blob_o_json = :json_blob
'''
# we prepare the *outer* statement that has a single command in it
statement = sql.text(outer_sql)
# now execute that passing in unicode value of the inner SQL.
logging.basicConfig(format='%(asctime)s %(message)s')
logging.getLogger().setLevel(logging.DEBUG)
nrows = [1000,10000,100000, 200000, 500000]
for n in nrows:
# we cons up a list of dicts to represent a single 'inline' table with lots of rows
logging.debug("passing %d items" % (n))
blob_o_json =json.dumps([dict(rn=x,foo=x*2,bar=x-1,fruit='banana', flavor='yummy') for x in range(0,n)])
# Note: we named arguments in this next line correspond to the :sql and :json_blob placeholders in outer_sql
result = engine.execute(statement,sql=six.text_type(inner_sql), json_blob=blob_o_json)
rs = result.fetchall()
#logging.debug((len(rs), rs[]))
# drop down to DBAPI to process each result-set. The application should know how many result-sets will be present
# I think it is likely that the vast majority of queries will just have a single result-set
# I did not know how to get to anything other than the first result-set ... I thought there
# was probably some pyodbc magic to make it work with SQLAlchemy and -- sure enough! -- there
# was a ticket with a workaround: https://github.com/sqlalchemy/sqlalchemy/issues/1635#issuecomment-441907416
#cursor = result.cursor
#rs1 = cursor.fetchall()
#cursor.nextset()
#rs2 = cursor.fetchall()
#logging.debug((len(rs1), rs1[-4:], rs2[-4:]))
logging.debug("finished %d" % (n))
logging.debug("done")
