Log raw SQL queries made by SQLAlchemy

sqlalchemy_sqlog.py

import time
from sqlalchemy import event

class sqlog:
    def __init__(self, session_or_engine):
        self.engine = (
            session_or_engine.get_bind()
            if hasattr(session_or_engine, "get_bind")
            else session_or_engine
        )

    def __enter__(self):
        event.listen(self.engine, "before_cursor_execute", self.before_exec)
        event.listen(self.engine, "after_cursor_execute", self.after_exec)

    def __exit__(self, *_, **__):
        event.remove(self.engine, "before_cursor_execute", self.before_exec)
        event.remove(self.engine, "after_cursor_execute", self.after_exec)

    def before_exec(self, *_, **__):
        self._ts = time.time()

    def after_exec(
        self, conn, cursor, statement, parameters, context, executemany
    ):
        statement = cursor.mogrify(statement, parameters).decode()
        print("SQL (%.4f): " % (time.time() - self._ts), statement)

  
"""
Usage:

with sqlog(session):
    b1 = Book()
    b1.id = 1
    b1.name = "HarryPotter"
    b1.author_id = jk.id
    session.add(b1)
    session.commit()

Output:

SQL (0.00106):  SELECT tbl_person.id AS tbl_person_id, tbl_person.name AS tbl_person_name 
FROM tbl_person 
WHERE tbl_person.id = 1
SQL (0.00154):  INSERT INTO tbl_book (id, name, author_id) VALUES (1, 'HarryPotter', 1)
"""
      

Extended version with pretty-print and syntax highlighting:

pip install pygments pygments-pprint-sql

import time
from sqlalchemy import event
from pygments import highlight
from pygments.lexers.sql import SqlLexer
from pygments.formatters import TerminalTrueColorFormatter
from pygments_pprint_sql import SqlFilter


formatter = TerminalTrueColorFormatter()
lexer = SqlLexer()
lexer.add_filter(SqlFilter())


class sqlog:
    def __init__(self, session_or_engine):
        self.engine = (
            session_or_engine.get_bind()
            if hasattr(session_or_engine, "get_bind")
            else session_or_engine
        )

    def __enter__(self):
        event.listen(self.engine, "before_cursor_execute", self.before_exec)
        event.listen(self.engine, "after_cursor_execute", self.after_exec)

    def __exit__(self, *_, **__):
        event.remove(self.engine, "before_cursor_execute", self.before_exec)
        event.remove(self.engine, "after_cursor_execute", self.after_exec)

    def before_exec(self, *_, **__):
        self._ts = time.time()

    def after_exec(
        self, conn, cursor, statement, parameters, context, executemany
    ):
        statement = cursor.mogrify(statement, parameters).decode()
        statement = highlight(statement, lexer, formatter)
        print(
            "\033[92m",
            "SQL (%.4f): " % (time.time() - self._ts),
            "\033[0m",
            statement,
        )