Debugging (Displaying) SQL Command Sent to the Db by SQLalchemy

Debugging (displaying) SQL command sent to the db by SQLAlchemy

In addition to echo parameter of create_engine() there is a more flexible way: configuring logging to echo engine statements:

import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

See Configuring Logging section of documentation for more information.

Flask SQLAlchemy display queries for debug

I haven't used it myself, but it seems that Flask Debug-toolbar may help with this.

https://github.com/mgood/flask-debugtoolbar

It's a port of the django-debug-toolbar, which can be used for profiling queries.
The documentation of Flask Debug-toolbar doesn't mention it, but there is code for a SQLAlchemyDebugPanel.

So I think it may be well worth to take a look at the project, and see if it does what you need.

How to log SQL statements and rows returned in sqlalchemy to aid in debugging?

Option 1: set the sqlalchemy.engine logger log level to either logging.INFO or logging.DEBUG:
e.g.,

>>> import logging
>>> logging.basicConfig()
>>> logger = logging.getLogger('sqlalchemy.engine')
>>> logger.setLevel(logging.DEBUG)
>>> session.query(User).all()
2015-01-02 11:54:25,854 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-01-02 11:54:25,856 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name
FROM users
2015-01-02 11:54:25,857 INFO sqlalchemy.engine.base.Engine {}
2015-01-02 11:54:25,858 DEBUG sqlalchemy.engine.base.Engine Col ('users_id', 'users_name')
2015-01-02 11:54:25,860 DEBUG sqlalchemy.engine.base.Engine Row (1, u'Alice')
2015-01-02 11:54:25,860 DEBUG sqlalchemy.engine.base.Engine Row (2, u'Bob')

Reference: Configuring Logging

Option 2: use the echo arg when calling sqlalchemy.create_engine():

e.g.,

>>> from sqlalchemy import create_engine
>>> from sqlalchemy.orm import sessionmaker
>>> engine = create_engine('postgres://postgres_user:my_password@localhost/my_db',
...                        echo="debug")
>>> Session = sessionmaker(bind=engine)
>>> session = Session()
>>> users = session.query(User).all()
2015-01-02 11:54:25,854 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-01-02 11:54:25,856 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name
FROM users
2015-01-02 11:54:25,857 INFO sqlalchemy.engine.base.Engine {}
2015-01-02 11:54:25,858 DEBUG sqlalchemy.engine.base.Engine Col ('users_id', 'users_name')
2015-01-02 11:54:25,860 DEBUG sqlalchemy.engine.base.Engine Row (1, u'Alice')
2015-01-02 11:54:25,860 DEBUG sqlalchemy.engine.base.Engine Row (2, u'Bob')

Per the sqlalchemy documentation:

*sqlalchemy.create_engine (*args, **kwargs)
...

echo=False – if True, the Engine will log all statements as well as a repr() of their parameter lists to the engines logger, which defaults to sys.stdout. The echo attribute of Engine can be modified at any time to turn logging on and off. If set to the string "debug", result rows will be printed to the standard output as well.

SQLAlchemy: print the actual query

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

Demo:

# 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

Show the SQL generated by Flask-SQLAlchemy

Flask-SQLAlchemy records debugging information about all queries during a request. You can get the information with get_debug_queries(). Flask-Debugtoolbar, for example, uses this to offer a debug panel with query timing information on each page render.

get_debug_queries() returns a list of queries in the order they were performed.

>>> from my_app import User
>>> from flask_sqlalchemy import get_debug_queries
>>> User.query.filter_by(display_name='davidism').all()
>>> info = get_debug_queries()[0]
>>> print(info.statement, info.parameters, info.duration, sep='\n')
SELECT "user".id AS user_id, se_user.id AS se_user_id, se_user.display_name AS se_user_display_name, se_user.profile_image AS se_user_profile_image, se_user.profile_link AS se_user_profile_link, se_user.reputation AS se_user_reputation, "user".superuser AS user_superuser \nFROM se_user JOIN "user" ON se_user.id = "user".id \nWHERE se_user.display_name = %(display_name_1)s
{'display_name_1': 'davidism'}
0.0016849040985107422

Queries are recorded if SQLALCHEMY_RECORD_QUERIES is set to True. This should be disabled when not needed for performance reasons.

---

Just calling str(query) does not show exactly what is sent to the database, as the final render is up to the lower-level database driver. SQLAlchemy has only the parameterized string and unescaped parameters. See the SQLAlchemy docs for a very detailed explanation of why. Usually it's good enough, but it's good to be aware that it's not the final query.

Enabling SQL statements for SQLAlchemy / Pyramid from the command line

If you're using pyramid_sqlalchemy, there's an option --sql-echo that you can use when running tests.

Otherwise, you will have parse the arguments yourself and set the logging level:

import logging
logging.getLogger('sqlalchemy.engine').setLevel('INFO')

For pserve and pshell, you would probably have to read the command line arguments somewhere in your app initalisation code.

For alembic, you would need to modify your env.py.

---

Related Topics