query_debugger.py

class QueryDebugger(object):
    def __init__(self, engine, query):
        with engine.connect() as connection:
            try:
                sqlalchemy.event.listen(engine, "do_execute", self.receive_do_execute)
                connection.execute(query)
            finally:
                sqlalchemy.event.remove(engine, "do_execute", self.receive_do_execute)

    def receive_do_execute(self, cursor, statement, parameters, context):
        self.statement = statement
        self.parameters = parameters
        self.query = cursor.mogrify(statement, parameters)
        # Don't actually execute
        return True

session.py

>>> import sqlalchemy
>>> engine = sqlalchemy.create_engine("postgresql://cureatr@localhost/cureatr_mongodb?sslmode=require")
>>> metadata = sqlalchemy.MetaData()
>>> users = sqlalchemy.Table('cureatr.users', metadata, sqlalchemy.Column("_id", sqlalchemy.String, primary_key=True), sqlalchemy.Column("document", sqlalchemy.dialects.postgresql.JSONB))
>>> s = sqlalchemy.select([users.c.document.label("foobar")]).where(users.c.document.contains({"profile": {"iid": "something"}}))
>>> q = QueryDebugger(engine, s)
>>> q.query
'SELECT "cureatr.users".document AS foobar \nFROM "cureatr.users" \nWHERE "cureatr.users".document @> \'{"profile": {"iid": "something"}}\''
>>> q.statement
'SELECT "cureatr.users".document AS foobar \nFROM "cureatr.users" \nWHERE "cureatr.users".document @> %(document_1)s'
>>> q.parameters
{'document_1': '{"profile": {"iid": "something"}}'}