From 8a6fd87d4b78b9837d0375c22034236633b2eeb0 Mon Sep 17 00:00:00 2001 From: Ryan Leckey Date: Fri, 26 Jul 2013 01:33:08 -0700 Subject: [PATCH] Add literal statement render utility. --- sqlalchemy_utils/__init__.py | 4 +++- sqlalchemy_utils/functions.py | 36 ++++++++++++++++++++++++++++++++- tests/test_utility_functions.py | 21 ++++++++++++++++++- 3 files changed, 58 insertions(+), 3 deletions(-) diff --git a/sqlalchemy_utils/__init__.py b/sqlalchemy_utils/__init__.py index fe034b1..8fc3426 100644 --- a/sqlalchemy_utils/__init__.py +++ b/sqlalchemy_utils/__init__.py @@ -1,6 +1,7 @@ from .exceptions import ImproperlyConfigured from .functions import ( - sort_query, defer_except, escape_like, primary_keys, table_name + sort_query, defer_except, escape_like, primary_keys, table_name, + render_statement ) from .listeners import coercion_listener from .merge import merge, Merger @@ -40,6 +41,7 @@ __all__ = ( merge, primary_keys, proxy_dict, + render_statement, table_name, ArrowType, ColorType, diff --git a/sqlalchemy_utils/functions.py b/sqlalchemy_utils/functions.py index 1446f8b..61a35f4 100644 --- a/sqlalchemy_utils/functions.py +++ b/sqlalchemy_utils/functions.py @@ -1,8 +1,10 @@ from collections import defaultdict +import six +import datetime import sqlalchemy as sa from sqlalchemy.orm import defer from sqlalchemy.orm.mapper import Mapper -from sqlalchemy.orm.query import _ColumnEntity +from sqlalchemy.orm.query import _ColumnEntity, Query from sqlalchemy.orm.properties import ColumnProperty from sqlalchemy.orm.util import AliasedInsp from sqlalchemy.schema import MetaData, Table, ForeignKeyConstraint @@ -352,3 +354,35 @@ def identity(obj): if column.primary_key: id_.append(getattr(obj, column.name)) return tuple(id_) + + +def render_statement(statement, bind=None): + """ + Generate an SQL expression string with bound parameters rendered inline + for the given SQLAlchemy statement. + """ + + if isinstance(statement, Query): + if bind is None: + bind = statement.session.get_bind(statement._mapper_zero_or_none()) + + statement = statement.statement + + elif bind is None: + bind = statement.bind + + class Compiler(bind.dialect.statement_compiler): + + def visit_bindparam(self, bindparam, *args, **kwargs): + return self.render_literal_value(bindparam.value, bindparam.type) + + def render_literal_value(self, value, type_): + if isinstance(value, six.integer_types): + return str(value) + + elif isinstance(value, (datetime.date, datetime.datetime)): + return "'%s'" % value + + return super(Compiler, self).render_literal_value(value, type_) + + return Compiler(bind.dialect, statement).process(statement) diff --git a/tests/test_utility_functions.py b/tests/test_utility_functions.py index 30f307f..6e845c0 100644 --- a/tests/test_utility_functions.py +++ b/tests/test_utility_functions.py @@ -1,7 +1,10 @@ import sqlalchemy as sa from sqlalchemy_utils import escape_like, defer_except -from sqlalchemy_utils.functions import non_indexed_foreign_keys from tests import TestCase +from sqlalchemy_utils.functions import ( + non_indexed_foreign_keys, + render_statement +) class TestEscapeLike(TestCase): @@ -62,3 +65,19 @@ class TestFindNonIndexedForeignKeys(TestCase): ] assert 'category_id' in column_names assert 'author_id' not in column_names + + def test_render_statement_query(self): + query = self.session.query(self.User).filter_by(id=3) + render = render_statement(query) + + assert 'SELECT user.id, user.name' in render + assert 'FROM user' in render + assert 'WHERE user.id = 3' in render + + def test_render_statement(self): + statement = self.User.__table__.select().where(self.User.id == 3) + render = render_statement(statement, bind=self.session.bind) + + assert 'SELECT user.id, user.name' in render + assert 'FROM user' in render + assert 'WHERE user.id = 3' in render