sqlalchemy-capture-sql

sqlalchemy-capture-sql 是一个库，可以捕获 SQLAlchemy SQL语句/查询。每个SQL语句都会捕获传递的参数和执行时间的大致值。它提供了报告和分析功能，例如：按SQL命令类型、表名、最慢查询列表等聚合。

动机

Django 有 django.db.connection.queries 连接属性，允许用户显示执行的原始SQL查询（仅DEBUG模式）。有时在调试或单元测试中，这可以用于检查和控制监视案例执行的SQL语句的数量和类型，以及允许的持续时间。我想为SQLAlchemy创建类似的功能，并提供额外的统计数据和分析函数。

工作原理

内部它使用 event.listens_for(engine, 'before_cursor_execute' 事件处理程序，例如。

@event.listens_for(engine, 'before_cursor_execute')
def capture_sa_statement_listener(...)

它简单地收集SQLAlchemy发送给事件监听器（在执行之前发送）的所有SQL语句，并在调用 .finish() 方法（或退出 "with" 上下文）之前将这些语句收集在 CaptureSqlStatements 实例中。

此外，它还提供时间测量（见备注）、统计和格式化函数，请参阅示例。

备注

一些备注

• 持续时间测量不是实际的DB执行时间，系统测量两个SQL语句捕获之间的时间

• 系统尝试检测SQL命令的类型（选择、插入、...）和首次引用的表/数据库对象名称，但背后的逻辑非常简单，不应依赖于它。

在Python 3.7+SQLAlchemy 1.3上进行了测试和开发，但我假设它应该可以在后续版本和一些早期版本上工作。

安装

像往常一样

pip install sqlalchemy-capture-sql

用法示例

标准用法是使用python的with语句

from sqlalchemy_capture_sql import CaptureSqlStatements

with CaptureSqlStatements(sqlalchemy_engine) as capture_stmts:

    # put here calls to functions that issue sqlalchemy commands that
    # produce some sql statements execution, for example factory-boy:
    cpm = FactoryModel.create()

    # call to .finish() automatically done on with ctx exit
capture_stmts.pp()

但标准风格也行 - 需要调用finish()方法

capture_stmts = CaptureSqlStatements(sqlalchemy_engine)

# put here calls to functions that issue sqlalchemy commands that
# produce some sql statements execution, for example factory-boy:
cpm = FactoryModel.create()

# in this case .finish() needs to be called to stop capturing
capture_stmts.finish()

调用美观打印函数

capture_stmts.pp()

库将生成完整报告，例如

============================================================
  1. 0.0020 INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
     <- 'joe+Joe Joey+joey'
  2. 0.0009 SELECT FROM users  WHERE users.id = ?
     <- '2'
  ...
============================================================
== Slowest (top 5):
      1. INSERT USERS             1   0.002 s INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
      2. INSERT USERS             1   0.001 s INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
      ...
============================================================
== By sql command (top 20):
    INSERT               2   0.003 s
    SELECT               4   0.003 s
    UPDATE               1   0.001 s
    DELETE               1   0.001 s
============================================================
== By table (top 20):
    USERS                6   0.007 s
    ...
============================================================
== By sql command + table (top 20):
    INSERT USERS             2   0.003 s
    SELECT USERS             2   0.002 s
    UPDATE USERS             1   0.001 s
    ...

== Totally captured 8 statement(s) in 0.008866 s

工作示例 - 一个较长的示例

此工作示例说明了某些原始SQL和ORM对象的用法（灵感来自SqlAlchemy 1.3教程）

from sqlalchemy_capture_sql import CaptureSqlStatements
from sqlalchemy import create_engine, text, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    nickname = Column(String)


engine = create_engine('sqlite:///:memory:', echo=False)
conn = engine.connect()

Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

session = Session()

# This orm operation won't be captured
user1 = User(name='ed', fullname='Ed Jones', nickname='edsnickname')
session.add(user1)
session.commit()

with CaptureSqlStatements(engine) as capture_stmts:
    # All commands within engine executed in this with block will be captured
    joe = User(name='joe', fullname='Joe Joey', nickname='joey')
    session.add(joe)
    session.commit()

    session.query(User).count()

    # one raw sql
    conn.execute(text("select 'In-capture'")).fetchall()

    joe.nickname = "Jo"
    session.commit()

    session.add(User(name='Wrong', fullname='Wrong', nickname='wrong'))
    session.rollback()

    jack = User(name='Jack', fullname='Jackson', nickname='jackie')
    session.add(jack)
    session.commit()

    session.delete(jack)
    session.commit()

# This orm operation won't be captured
session.add(User(name='Mick', fullname='Michael', nickname='mick'))
assert session.query(User).count(), 3

assert capture_stmts.get_counts("by_type"), {'INSERT': 2, 'SELECT': 4, 'UPDATE': 1, 'DELETE': 1}
assert capture_stmts.get_counts("by_table"), {"'IN-CAPTURE'": 1, '(SELECT': 1, 'USERS': 6}
assert capture_stmts.get_counts("by_type_and_table"), {
    'DELETE USERS': 1,
    'INSERT USERS': 2,
    "SELECT 'IN-CAPTURE'": 1,
    'SELECT (SELECT': 1,
    'SELECT USERS': 2,
    'UPDATE USERS': 1}

assert [st.stmt_repr for st in capture_stmts.statements], [
    'INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)',
    'SELECT FROM (SELECT users.id AS users_id, users.name AS users_name, '
        'users.fullname AS users_fullname, users.nickname AS users_nickname \n'
        'FROM users) AS anon_1',
    "select 'In-capture'",
    'SELECT FROM users \nWHERE users.id = ?',
    'UPDATE users SET nickname=? WHERE users.id = ?',
    'INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)',
    'SELECT FROM users \nWHERE users.id = ?',
    'DELETE FROM users WHERE users.id = ?']

调用美观打印函数

capture_stmts.pp()

产生

============================================================
== NOTE: duration measures time between 2 captures, it is not actual DB execution time.
== Totally captured 8 statement(s) in 0.008866 s:
  1. 0.0020 INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
     <- 'joe+Joe Joey+joey'
  2. 0.0005 SELECT FROM (SELECT users.id AS users_id, users.name AS users_name, us
  3. 0.0010 select 'In-capture'
  4. 0.0009 SELECT FROM users  WHERE users.id = ?
     <- '2'
  5. 0.0013 UPDATE users SET nickname=? WHERE users.id = ?
     <- 'Jo+2'
  6. 0.0014 INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
     <- 'Jack+Jackson+jackie'
  7. 0.0007 SELECT FROM users  WHERE users.id = ?
     <- '3'
  8. 0.0005 DELETE FROM users WHERE users.id = ?
     <- '3'

============================================================
== Slowest (top 5):
      1. INSERT USERS             1   0.002 s INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
      2. INSERT USERS             1   0.001 s INSERT INTO users (name, fullname, nickname) VALUES (?, ?, ?)
      3. UPDATE USERS             1   0.001 s UPDATE users SET nickname=? WHERE users.id = ?
      4. SELECT 'IN-CAPTURE'      1   0.001 s select 'In-capture'
      5. SELECT USERS             1   0.001 s SELECT FROM users
WHERE users.id = ?

============================================================
== By sql command (top 20):
    INSERT               2   0.003 s
    SELECT               4   0.003 s
    UPDATE               1   0.001 s
    DELETE               1   0.001 s

============================================================
== By table (top 20):
    USERS                6   0.007 s
    'IN-CAPTURE'         1   0.001 s
    (SELECT              1   0.000 s

============================================================
== By sql command + table (top 20):
    INSERT USERS             2   0.003 s
    SELECT USERS             2   0.002 s
    UPDATE USERS             1   0.001 s
    SELECT 'IN-CAPTURE'      1   0.001 s
    DELETE USERS             1   0.001 s
    SELECT (SELECT           1   0.000 s

============================================================
== Totally captured 8 statement(s) in 0.008866 s

可以迭代所有捕获语句对象

for statement in capture_stmts:
    print(statement.statement)
    print(statement.tst_started)
    print(statement.duration)    # BEWARE: not actual DB execution time, 
                                           Rounded on 2 decimal places.
    print(statement.stmt_repr)   # Dropped list of columns from SELECT
    print(statement.parameters)
    print(statement.executemany) # bool
    print(statement.sql_type)    # BEWARE: do not rely on this
    print(statement.first_table) # BEWARE: do not rely on this

杂项

其他方法

检查CaptureSqlStatements的其他实例方法，以获取列表/字典对象中的统计信息，例如

count() -> int
get_counts(name:StatName) -> Dict[str, int]
get_slowest(top:int=TOP_DEFAULT_SLOWEST) -> List[Stat]
get_statement_by_row_id(row_id:int) -> SqlStatement
get_stats(name: StatName, top:int=TOP_DEFAULT) -> Tuple[int, List[Stat]]
pp(verbose:bool=False, print_cmd:Callable=print)
report_counter(name: StatName, top:int=TOP_DEFAULT) -> str
report_slowest(verbose=False) -> str
report_stats(name: StatName, top:int=TOP_DEFAULT, fields:List[str]=AGG_FIELDS) -> str

SQLite3内部数据库

库内部使用sqlite3内存数据库来存储基本语句信息。数据库用于统计、聚合和查找最慢的查询，但用户也可以用它进行进一步分析，例如

cursor = capture_stmts.connection.cursor()
cursor.execute(f"select id, sql_type, first_table, duration from sql_statement order by duration desc limit 100")
for row in cursor.fetchall():
    row_id, sql_type, first_table, duration = row
    stmt = capture_stmts.get_statement_by_row_id(row_id)
    print(f"{sql_type} {first_table} {duration} : {stmt.statement} <- {stmt.parameters}")

SQLAlchemy语句记录

当SQLAlchemy的“sqlalchemy.engine”日志级别设置为INFO级别或更高时，SQLAlchemy可以记录SQL语句

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

此功能也已作为CaptureSqlStatements的静态方法附加

sqlalchemy_log_statements_enable()
sqlalchemy_log_statements_disable()

另一种选择是在创建具有echo属性的SqlAlchemy.Engine时提供echo=True属性，例如

engine = create_engine('sqlite:///...', echo=True)

运行测试

克隆仓库，进入根目录并运行

python tests/test_basic.py

要查看详细输出 - pp()结果，请按如下方式运行

VERBOSE=1 python tests/test_basic.py