python - 我如何获得查询执行时间的 psycopg2 日志记录？

Question

我正在尝试获取有关 psycopg2 执行的查询的性能统计信息，但文档/示例似乎仍然模糊不清，并没有达到应有的清晰度。

我至少可以通过记录器进行调试。 我需要做什么才能访问查询的性能数据？我想要获取查询执行时间的数字。

有没有我可以访问的方法，或者我需要初始化的其他东西来输出查询执行时间？

这是我目前所掌握的内容的拼凑摘录:

import psycopg2
import psycopg2.extensions
from psycopg2.extras import LoggingConnection
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

# set higher up in script
db_settings = {
    "user": user,
    "password": password,
    "host": host,
    "database": dbname,
}

query_txt = "[query_txt_from file]"

conn = psycopg2.connect(connection_factory=LoggingConnection, **db_settings)
conn.initialize(logger)

cur = conn.cursor()
cur.execute(query_txt)

我明白了

DEBUG:__main__: [the query executed]

Answer 1

很容易在执行开始时设置时间戳并在结束时计算持续时间。您将需要您自己的 LoggingConnection 和 LoggingCursor 的简单子(monad)类。请参阅我的示例代码。

这基于您可以在 psycopg2/extras.py 源中找到的 MinTimeLoggingConnection 源。

import time
import psycopg2
import psycopg2.extensions
from psycopg2.extras import LoggingConnection, LoggingCursor
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

# MyLoggingCursor simply sets self.timestamp at start of each query                                                                 
class MyLoggingCursor(LoggingCursor):
    def execute(self, query, vars=None):
        self.timestamp = time.time()
        return super(MyLoggingCursor, self).execute(query, vars)

    def callproc(self, procname, vars=None):
        self.timestamp = time.time()
        return super(MyLoggingCursor, self).callproc(procname, vars)

# MyLogging Connection:                                                                                                             
#   a) calls MyLoggingCursor rather than the default                                                                                
#   b) adds resulting execution (+ transport) time via filter()                                                                     
class MyLoggingConnection(LoggingConnection):
    def filter(self, msg, curs):
        return msg + "   %d ms" % int((time.time() - curs.timestamp) * 1000)

    def cursor(self, *args, **kwargs):
        kwargs.setdefault('cursor_factory', MyLoggingCursor)
        return LoggingConnection.cursor(self, *args, **kwargs)

db_settings = {
    ....
}

query_txt = "[query_text_from file]"

conn = psycopg2.connect(connection_factory=MyLoggingConnection, **db_settings)
conn.initialize(logger)

cur = conn.cursor()
cur.execute(query_text)

你会得到:

DEBUG: __main__:[query]     3 ms

在您的 filter() 中，您可以更改格式，或者如果小于某个值则选择不显示。