Skip to main content
update for 1.0
Source Link
bukzor
  • 38.8k
  • 13
  • 85
  • 116

Combining the existing recipesThis works in python 2 and 3 and is a bit cleaner than before, but using inheritance rather than monkey-patching:requires SA>=1.0.

deffrom literalquery(statement,sqlalchemy.engine.default dialect=None):
import DefaultDialect
from sqlalchemy.sql.sqltypes import """GenerateString, anDateTime, SQLNullType

# expressionpython2/3 stringcompatible.
PY3 with= boundstr parametersis renderednot inlinebytes
  text = forstr theif givenPY3 SQLAlchemyelse statement.
unicode
    WARNING: Thisint_type method= ofint escapingif isPY3 insecure,else incomplete(int, and for debugginglong)
    purposes only. Executingstr_type SQL= statementsstr withif inline-renderedPY3 userelse values(str, isunicode)
    extremely

class insecure.StringLiteral(String):
    """
"""Teach SA how to importliteralize sqlalchemyvarious things.orm"""
    ifdef isinstanceliteral_processor(statementself, sqlalchemy.orm.Querydialect):
        ifsuper_processor dialect= issuper(StringLiteral, None:self).literal_processor(dialect)
      
       dialect =def statement.session.get_bindprocess(value):
               if statement._mapper_zero_or_noneisinstance(value, int_type):
            ).dialect
        statement =return statement.statementtext(value)
    if dialect is None:
        dialectif =not getattrisinstance(statement.bind, 'dialect'value, Nonestr_type)
    if dialect is None:
        from sqlalchemy.dialects import mysql
        dialect = mysql.dialect()

    Compilervalue = type(statement._compilertext(dialect)value)
 
    class LiteralCompiler(Compiler):
        visit_bindparamresult = Compiler.render_literal_bindparam

        def render_literal_valuesuper_processor(self, value, type_):
            if isinstance(value, (Decimalresult, long)bytes):
                returnresult str= result.decode(valuedialect.encoding)
            elif isinstance(value,return datetime):result
                return repr(strprocess


class LiteralDialect(value)DefaultDialect):
    colspecs = {
      else:  # fallback
   prevent various encoding explosions
        String: StringLiteral,
 value = super(LiteralCompiler, self).render_literal_value(
    # teach SA about how to literalize a datetime
        value,DateTime: type_StringLiteral,
         # don't format py2 long integers to )NULL
           NullType: StringLiteral,
    if}


def isinstanceliteralquery(value, unicodestatement):
      """NOTE: This is entirely insecure. DO NOT execute the resulting strings."""
    returnimport valuesqlalchemy.encode('UTF-8')orm
    if isinstance(statement, sqlalchemy.orm.Query):
        statement = else:statement.statement
    return statement.compile(
        dialect=LiteralDialect(),
       return value

compile_kwargs={'literal_binds': True},
   return LiteralCompiler(dialect, statement).string
# coding: UTF-8
from sqlalchemy.dialects import mysql
from datetime import datetime
from decimal import Decimal

from literalquery import literalquery


def test():
    from sqlalchemy.sql import table, column, select

    mytable = table('mytable', column('mycol'))
    values = (
        5,
        u'snowman: ☃',
        b'UTF-8 snowman: \xe2\x98\x83',
        datetime.now(),
        Decimal('3.14159'),
        long(1234)10 ** 20,  # a long integer
    )

    statement = select([mytable]).where(mytable.c.mycol.in_(values)).limit(1)
    print (literalquery(statement))


if __name__ == '__main__':
    test()

Gives this output: (tested in python 2.7 and 3.4)

SELECT mytable.mycol
FROM mytable
WHERE mytable.mycol IN (5, 'snowman: ☃', 'UTF-8 snowman: ☃',
    '2014  '2015-0806-0124 1118:1909:5429.514764'042517', 3.14159, 1234100000000000000000000)
 LIMIT 1

Combining the existing recipes, but using inheritance rather than monkey-patching:

def literalquery(statement, dialect=None):
    """Generate an SQL expression string with bound parameters rendered inline
    for the given SQLAlchemy statement.

    WARNING: This method of escaping is insecure, incomplete, and for debugging
    purposes only. Executing SQL statements with inline-rendered user values is
    extremely insecure.
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        if dialect is None:
            dialect = statement.session.get_bind(
                statement._mapper_zero_or_none()
            ).dialect
        statement = statement.statement
    if dialect is None:
        dialect = getattr(statement.bind, 'dialect', None)
    if dialect is None:
        from sqlalchemy.dialects import mysql
        dialect = mysql.dialect()

    Compiler = type(statement._compiler(dialect))
 
    class LiteralCompiler(Compiler):
        visit_bindparam = Compiler.render_literal_bindparam

        def render_literal_value(self, value, type_):
            if isinstance(value, (Decimal, long)):
                return str(value)
            elif isinstance(value, datetime):
                return repr(str(value))
            else:  # fallback
                value = super(LiteralCompiler, self).render_literal_value(
                    value, type_,
                )
                if isinstance(value, unicode):
                    return value.encode('UTF-8')
                else:
                    return value

    return LiteralCompiler(dialect, statement)
# coding: UTF-8
from sqlalchemy.dialects import mysql
from datetime import datetime
from decimal import Decimal

from literalquery import literalquery


def test():
    from sqlalchemy.sql import table, column, select

    mytable = table('mytable', column('mycol'))
    values = (
        5,
        u'snowman: ☃',
        b'UTF-8 snowman: \xe2\x98\x83',
        datetime.now(),
        Decimal('3.14159'),
        long(1234),
    )

    statement = select([mytable]).where(mytable.c.mycol.in_(values)).limit(1)
    print literalquery(statement)


if __name__ == '__main__':
    test()

Gives this output:

SELECT mytable.mycol
FROM mytable
WHERE mytable.mycol IN (5, 'snowman: ☃', 'UTF-8 snowman: ☃',
    '2014-08-01 11:19:54.514764', 3.14159, 1234)
 LIMIT 1

This works in python 2 and 3 and is a bit cleaner than before, but requires SA>=1.0.

from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.sql.sqltypes import String, DateTime, NullType

# python2/3 compatible.
PY3 = str is not bytes
text = str if PY3 else unicode
int_type = int if PY3 else (int, long)
str_type = str if PY3 else (str, unicode)


class StringLiteral(String):
    """Teach SA how to literalize various things."""
    def literal_processor(self, dialect):
        super_processor = super(StringLiteral, self).literal_processor(dialect)
 
        def process(value):
            if isinstance(value, int_type):
                return text(value)
            if not isinstance(value, str_type):
                value = text(value)
            result = super_processor(value)
            if isinstance(result, bytes):
                result = result.decode(dialect.encoding)
            return result
        return process


class LiteralDialect(DefaultDialect):
    colspecs = {
        # prevent various encoding explosions
        String: StringLiteral,
        # teach SA about how to literalize a datetime
        DateTime: StringLiteral,
        # don't format py2 long integers to NULL
        NullType: StringLiteral,
    }


def literalquery(statement):
    """NOTE: This is entirely insecure. DO NOT execute the resulting strings."""
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        statement = statement.statement
    return statement.compile(
        dialect=LiteralDialect(),
        compile_kwargs={'literal_binds': True},
    ).string
# coding: UTF-8
from datetime import datetime
from decimal import Decimal

from literalquery import literalquery


def test():
    from sqlalchemy.sql import table, column, select

    mytable = table('mytable', column('mycol'))
    values = (
        5,
        u'snowman: ☃',
        b'UTF-8 snowman: \xe2\x98\x83',
        datetime.now(),
        Decimal('3.14159'),
        10 ** 20,  # a long integer
    )

    statement = select([mytable]).where(mytable.c.mycol.in_(values)).limit(1)
    print(literalquery(statement))


if __name__ == '__main__':
    test()

Gives this output: (tested in python 2.7 and 3.4)

SELECT mytable.mycol
FROM mytable
WHERE mytable.mycol IN (5, 'snowman: ☃', 'UTF-8 snowman: ☃',
      '2015-06-24 18:09:29.042517', 3.14159, 100000000000000000000)
 LIMIT 1
stole good ideas from @gsakkis and @vvladymyrov
Source Link
bukzor
  • 38.8k
  • 13
  • 85
  • 116
def printqueryliteralquery(statement, bind=Nonedialect=None):
    """
  """Generate an printSQL aexpression query,string with valuesbound filledparameters inrendered inline
    for debuggingthe purposesgiven *only*SQLAlchemy statement.

    forWARNING: security,This youmethod shouldof alwaysescaping separateis queriesinsecure, fromincomplete, theirand valuesfor debugging
    pleasepurposes alsoonly. noteExecuting thatSQL thisstatements functionwith inline-rendered user values is 
 quite slow  extremely insecure.
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        if binddialect is None:
            binddialect = statement.session.get_bind(
                    statement._mapper_zero_or_none()
            ).dialect
        statement = statement.statement
    elifif binddialect is None:
        binddialect = getattr(statement.bind, 'dialect', None)
    if dialect is None:
        from sqlalchemy.dialects import mysql
        dialect = bindmysql.dialect()

    compilerCompiler = type(statement._compiler(dialect)) 

    class LiteralCompiler(compiler.__class__Compiler):
        def visit_bindparam( = Compiler.render_literal_bindparam

        def render_literal_value(self, value, type_):
          self, bindparam if isinstance(value, within_columns_clause=False(Decimal, long)):
                literal_binds=False,return **kwargsstr(value)
            elif isinstance(value, datetime):
                return repr(str(value))
            else:  # fallback
                value = super(LiteralCompiler, self).render_literal_bindparamrender_literal_value(
                    bindparamvalue, within_columns_clause=within_columns_clausetype_,
                )
    literal_binds=literal_binds            if isinstance(value, **kwargsunicode):
                    return value.encode('UTF-8')
                else:
    compiler =               return value

    return LiteralCompiler(dialect, statement)

Demo:

# coding: UTF-8
from sqlalchemy.dialects printimport compilermysql
from datetime import datetime
from decimal import Decimal

from literalquery import literalquery


def test():
    from sqlalchemy.processsql import table, column, select

    mytable = table('mytable', column('mycol'))
    values = (
        5,
        u'snowman: ☃',
        b'UTF-8 snowman: \xe2\x98\x83',
        datetime.now(),
        Decimal('3.14159'),
        long(1234),
    )

    statement = select([mytable]).where(mytable.c.mycol.in_(values)).limit(1)
    print literalquery(statement)


if __name__ == '__main__':
    test()

Gives this output:

SELECT mytable.mycol
FROM mytable
WHERE mytable.mycol IN (5, 'snowman: ☃', 'UTF-8 snowman: ☃',
    '2014-08-01 11:19:54.514764', 3.14159, 1234)
 LIMIT 1
def printquery(statement, bind=None):
    """
    print a query, with values filled in
    for debugging purposes *only*
    for security, you should always separate queries from their values
    please also note that this function is quite slow
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.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 

    dialect = bind.dialect
    compiler = statement._compiler(dialect)
    class LiteralCompiler(compiler.__class__):
        def visit_bindparam(
                self, bindparam, within_columns_clause=False, 
                literal_binds=False, **kwargs
        ):
            return super(LiteralCompiler, self).render_literal_bindparam(
                    bindparam, within_columns_clause=within_columns_clause,
                    literal_binds=literal_binds, **kwargs
            )

    compiler = LiteralCompiler(dialect, statement)
    print compiler.process(statement)
def literalquery(statement, dialect=None):
    """Generate an SQL expression string with bound parameters rendered inline
    for the given SQLAlchemy statement.

    WARNING: This method of escaping is insecure, incomplete, and for debugging
    purposes only. Executing SQL statements with inline-rendered user values is 
    extremely insecure.
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        if dialect is None:
            dialect = statement.session.get_bind(
                statement._mapper_zero_or_none()
            ).dialect
        statement = statement.statement
    if dialect is None:
        dialect = getattr(statement.bind, 'dialect', None)
    if dialect is None:
        from sqlalchemy.dialects import mysql
        dialect = mysql.dialect()

    Compiler = type(statement._compiler(dialect)) 

    class LiteralCompiler(Compiler):
        visit_bindparam = Compiler.render_literal_bindparam

        def render_literal_value(self, value, type_):
            if isinstance(value, (Decimal, long)):
                return str(value)
            elif isinstance(value, datetime):
                return repr(str(value))
            else:  # fallback
                value = super(LiteralCompiler, self).render_literal_value(
                    value, type_,
                )
                if isinstance(value, unicode):
                    return value.encode('UTF-8')
                else:
                    return value

    return LiteralCompiler(dialect, statement)

Demo:

# coding: UTF-8
from sqlalchemy.dialects import mysql
from datetime import datetime
from decimal import Decimal

from literalquery import literalquery


def test():
    from sqlalchemy.sql import table, column, select

    mytable = table('mytable', column('mycol'))
    values = (
        5,
        u'snowman: ☃',
        b'UTF-8 snowman: \xe2\x98\x83',
        datetime.now(),
        Decimal('3.14159'),
        long(1234),
    )

    statement = select([mytable]).where(mytable.c.mycol.in_(values)).limit(1)
    print literalquery(statement)


if __name__ == '__main__':
    test()

Gives this output:

SELECT mytable.mycol
FROM mytable
WHERE mytable.mycol IN (5, 'snowman: ☃', 'UTF-8 snowman: ☃',
    '2014-08-01 11:19:54.514764', 3.14159, 1234)
 LIMIT 1
deleted 4 characters in body
Source Link
bukzor
  • 38.8k
  • 13
  • 85
  • 116

Combining the existing recipes, but using inheritance rather than monkey-patching:

def printquery(statement, bind=None):
    """
    print a query, with values filled in
    for debugging purposes *only*
    for security, you should always separate queries from their values
    please also note that this function is quite slow
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.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 

    dialect = bind.dialect
    compiler = statement._compiler(dialect)
    class LiteralCompiler(compiler.__class__):
        def visit_bindparam(
                self, bindparam, within_columns_clause=False, 
                literal_binds=False, **kwargs
        ):
            return super(LiteralCompiler, self).render_literal_bindparam(
                    bindparam, within_columns_clause=within_columns_clause,
                    literal_binds=literal_binds, **kwargs
            )

    compiler = LiteralCompiler(dialect, statement)
    print compiler.process(statement)

Combining the existing recipes, but using inheritance rather than monkey-patching:

def printquery(statement, bind=None):
    """
    print a query, with values filled in
    for debugging purposes *only*
    for security, you should always separate queries from their values
    please also note that this function is quite slow
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.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 

    dialect = bind.dialect
    compiler = statement._compiler(dialect)
    class LiteralCompiler(compiler.__class__):
        def visit_bindparam(
                self, bindparam, within_columns_clause=False, 
                literal_binds=False, **kwargs
        ):
            return super(LiteralCompiler, self).render_literal_bindparam(
                    bindparam, within_columns_clause=within_columns_clause,
                    literal_binds=literal_binds, **kwargs
            )

    compiler = LiteralCompiler(dialect, statement)
    print compiler.process(statement)

Combining the existing recipes, but using inheritance rather than monkey-patching:

def printquery(statement, bind=None):
    """
    print a query, with values filled in
    for debugging purposes *only*
    for security, you should always separate queries from their values
    please also note that this function is quite slow
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.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 

    dialect = bind.dialect
    compiler = statement._compiler(dialect)
    class LiteralCompiler(compiler.__class__):
        def visit_bindparam(
                self, bindparam, within_columns_clause=False, 
                literal_binds=False, **kwargs
        ):
            return super(LiteralCompiler, self).render_literal_bindparam(
                    bindparam, within_columns_clause=within_columns_clause,
                    literal_binds=literal_binds, **kwargs
            )

    compiler = LiteralCompiler(dialect, statement)
    print compiler.process(statement)
Source Link
bukzor
  • 38.8k
  • 13
  • 85
  • 116
Loading