Merge pull request #30 from kvesteri/topics/render-statement

Add literal statement render utility.

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
@@ -41,6 +42,7 @@ __all__ = (
     merge,
     primary_keys,
     proxy_dict,
+    render_statement,
     table_name,
     ArrowType,
     ColorType,

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)

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