/usr/lib/python2.7/dist-packages/sqlalchemy_utils/functions/render.py is in python-sqlalchemy-utils 0.30.12-4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | import inspect
import six
import sqlalchemy as sa
from .mock import create_mock_engine
def render_expression(expression, bind, stream=None):
"""Generate a SQL expression from the passed python expression.
Only the global variable, `engine`, is available for use in the
expression. Additional local variables may be passed in the context
parameter.
Note this function is meant for convenience and protected usage. Do NOT
blindly pass user input to this function as it uses exec.
:param bind: A SQLAlchemy engine or bind URL.
:param stream: Render all DDL operations to the stream.
"""
# Create a stream if not present.
if stream is None:
stream = six.moves.cStringIO()
engine = create_mock_engine(bind, stream)
# Navigate the stack and find the calling frame that allows the
# expression to execuate.
for frame in inspect.stack()[1:]:
try:
frame = frame[0]
local = dict(frame.f_locals)
local['engine'] = engine
six.exec_(expression, frame.f_globals, local)
break
except:
pass
else:
raise ValueError('Not a valid python expression', engine)
return stream
def render_statement(statement, bind=None):
"""
Generate an SQL expression string with bound parameters rendered inline
for the given SQLAlchemy statement.
:param statement: SQLAlchemy Query object.
:param bind:
Optional SQLAlchemy bind, if None uses the bind of the given query
object.
"""
if isinstance(statement, sa.orm.query.Query):
if bind is None:
bind = statement.session.get_bind(statement._mapper_zero())
statement = statement.statement
elif bind is None:
bind = statement.bind
stream = six.moves.cStringIO()
engine = create_mock_engine(bind.engine, stream=stream)
engine.execute(statement)
return stream.getvalue()
|