snapshot-queries

捕获通过Django和SqlAlchemy ORM查询执行的SQL语句

使用场景

• 查看由复杂的Django Queryset或SQLAlchemy查询执行的精确查询
• 快速查看代码块中执行的慢查询
• 识别代码块中执行的类似查询，以帮助检测N+1查询问题
• 查看触发查询执行的精确代码行，包括其完整的堆栈跟踪

示例

• 显示代码块中执行的查询
• 显示每个查询的特定属性
• 按持续时间排序查询
• 检查最慢的查询
• 将具有重复SQL语句的查询分组在一起
• 将具有类似SQL语句的查询分组在一起

显示代码块中执行的查询

from django.contrib.auth import get_user_model
from snapshot_queries import snapshot_queries

User = get_user_model()
with snapshot_queries() as queries_executed:
    User.objects.only('email').get(id=1)
    User.objects.only('email').get(id=7)

queries_executed.display()

输出

Query 1
---------
2 ms

/path/to/module.py:5 in function_name

User.objects.only('email').get(id=1)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 1


Query 2
---------
< 1 ms

/path/to/module.py:6 in function_name

User.objects.only('email').get(id=7)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 7

显示每个查询的特定属性

您可以选择显示哪些属性。

支持的显示属性

• code (触发查询的Python代码)
• duration (查询执行所需的时间)
• idx (执行查询的索引)
• location (查询执行的代码位置)
• stacktrace (每个查询的完整堆栈跟踪)
• sql (查询的SQL语句)
• colored (显示着色的SQL语句)
• formatted (显示格式化的SQL语句)

例如，要显示每个查询的堆栈跟踪，请使用 queries_executed.display(stacktrace=True)

from django.contrib.auth import get_user_model
from snapshot_queries import snapshot_queries

User = get_user_model()

def main():
    with snapshot_queries() as queries_executed:
        User.objects.only('email').get(id=1)
        User.objects.only('is_staff').get(id=7)

    queries_executed.display(sql=True, stacktrace=True)

main()

输出

Query 1
---------
./path/to/file.py:12 in <module>
    main()
./path/to/file.py:8 in main
    User.objects.only('email').get(id=1)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 1


Query 2
---------
./path/to/file.py:13 in <module>
    main()
./path/to/file.py:9 in main
    User.objects.only('is_staff').get(id=7)

SELECT "auth_user"."id",
       "auth_user"."is_staff"
FROM "auth_user"
WHERE "auth_user"."id" = 7

按持续时间排序查询

fastest_queries = queries_executed.order_by('duration')[:3]
slowest_queries = queries_executed.order_by('-duration')[:3]
slowest_queries_executed.display()

检查最慢的查询

slowest_query = queries_executed.order_by('-duration')[0]
slowest_query.display(code=True, location=True, sql=True)

将具有重复SQL语句的查询分组在一起

from django.contrib.auth import get_user_model
from snapshot_queries import snapshot_queries

User = get_user_model()

with snapshot_queries() as queries_executed:
    User.objects.only('email').get(id=1)
    User.objects.only('email').get(id=1)
    User.objects.only('email').get(id=1)
    User.objects.only('email').get(id=7)

duplicates = queries_executed.duplicates().display()

输出

========================
3 duplicate queries
========================
Query 1
---------
1 ms

./path/to/file.py:9 in main

User.objects.only('email').get(id=1)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 1


Query 2
---------
< 1 ms

./path/to/file.py:10 in main

User.objects.only('email').get(id=1)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 1


Query 3
---------
< 1 ms

./path/to/file.py:11 in main

User.objects.only('email').get(id=1)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 1

将具有类似SQL语句的查询分组在一起

from django.contrib.auth import get_user_model
from snapshot_queries import snapshot_queries

User = get_user_model()

with snapshot_queries() as queries_executed:
    User.objects.only('email').get(id=1)
    User.objects.only('email').get(id=1)
    User.objects.only('email').get(id=1)
    User.objects.only('email').get(id=7)

similar = queries_executed.similar().display()

输出

========================
4 similar queries
========================
Query 1
---------
2 ms

/path/to/file.py:6 in main

User.objects.only('email').get(id=1)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 1


Query 2
---------
< 1 ms

/path/to/file.py:7 in main

User.objects.only('email').get(id=1)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 1


Query 3
---------
< 1 ms

/path/to/file.py:8 in main

User.objects.only('email').get(id=1)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 1


Query 4
---------
< 1 ms

/path/to/file.py:9 in main

User.objects.only('email').get(id=7)

SELECT "auth_user"."id",
       "auth_user"."email"
FROM "auth_user"
WHERE "auth_user"."id" = 7