Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import inspect
2import io
4import sqlalchemy as sa
6from .mock import create_mock_engine
7from .orm import _get_query_compile_state
10def render_expression(expression, bind, stream=None):
11 """Generate a SQL expression from the passed python expression.
13 Only the global variable, `engine`, is available for use in the
14 expression. Additional local variables may be passed in the context
15 parameter.
17 Note this function is meant for convenience and protected usage. Do NOT
18 blindly pass user input to this function as it uses exec.
20 :param bind: A SQLAlchemy engine or bind URL.
21 :param stream: Render all DDL operations to the stream.
22 """
24 # Create a stream if not present.
26 if stream is None:
27 stream = io.StringIO()
29 engine = create_mock_engine(bind, stream)
31 # Navigate the stack and find the calling frame that allows the
32 # expression to execuate.
34 for frame in inspect.stack()[1:]:
35 try:
36 frame = frame[0]
37 local = dict(frame.f_locals)
38 local['engine'] = engine
39 exec(expression, frame.f_globals, local)
40 break
41 except Exception:
42 pass
43 else:
44 raise ValueError('Not a valid python expression', engine)
46 return stream
49def render_statement(statement, bind=None):
50 """
51 Generate an SQL expression string with bound parameters rendered inline
52 for the given SQLAlchemy statement.
54 :param statement: SQLAlchemy Query object.
55 :param bind:
56 Optional SQLAlchemy bind, if None uses the bind of the given query
57 object.
58 """
60 if isinstance(statement, sa.orm.query.Query):
61 if bind is None:
62 bind = statement.session.get_bind(
63 _get_query_compile_state(statement)._mapper_zero()
64 )
66 statement = statement.statement
68 elif bind is None:
69 bind = statement.bind
71 stream = io.StringIO()
72 engine = create_mock_engine(bind.engine, stream=stream)
73 engine.execute(statement)
75 return stream.getvalue()