SQLAlchemy debug utils

sqlalchemy_debug_utils.py

def to_sql(query, dialect_name=None):
    from sqlalchemy import dialects
    import operator
    
    if dialect_name:
        dialect_module = getattr(dialects, dialect_name)
        if not dialect_module:
            raise KeyError('Wrong dialect {}'.format(dialect_name))
        dialect = dialect_module.dialect()
    else:
        try:
            # think about it as getattr(query, 'session.bind.dialect')
            dialect = operator.attrgetter('session.bind.dialect')(query)
        except AttributeError:
            raise KeyError('dialect_name was not passed '
                           'and cant get dialect from session.'
                           'Pass dialect_name or bound query to session')
    return str(
        query.statement.compile(compile_kwargs={"literal_binds": True},
                                dialect=dialect))


def register_sqlalchemy_echo_with_bound_params():
    """
    it's like
        create_engine(echo=True),
    but binds params to query, so you have real raw SQL
    """
    import sys
    from datetime import datetime
    from sqlalchemy.engine import Engine
    from sqlalchemy import event

    @event.listens_for(Engine, "before_cursor_execute")
    def before_cursor_execute(conn, cursor, statement,
                              parameters, context, executemany):
        # wrap non-integers in quotes
        if isinstance(parameters, dict):
            escaped_params = {}
            for k, v in parameters.items():
                escaped_params[k] = v if isinstance(v, (int, long)) \
                                    else "'{}'".format(v)
            sys.stdout.write("\n{} QUERY: \n{}\n".format(
                datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                statement % escaped_params))
        else:
            sys.stdout.write("\n{} QUERY: \n{}\nParams: {}\n".format(
                datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
                statement, parameters))