Python psycopg2 - 记录事件答案

【问题标题】：Python psycopg2 - Logging eventsPython psycopg2 - 记录事件
【发布时间】：2012-12-30 02:40:56
【问题描述】：

我正在使用 psycopg2，但在将事件（执行的查询、通知、错误）记录到文件时遇到问题。我想像在 PgAdmin 历史记录窗口中一样生效。

例如我正在执行这个查询：

insert into city(id, name, countrycode, district, population) values (4080,'Savilla', 'ESP', 'andalucia', 1000000)

在 PgAdmin 中我看到这样的效果：

Executing query: 
insert into city(id, name, countrycode, district, population) values (4080,'Sevilla', 'ESP', 'andalucia', 1000000)

Query executed in 26 ms.
One row affected.

我可以使用 psycopg2 获得类似的效果吗？

我尝试使用LoggingCursor，但对我来说并不满意，因为它只记录查询。

感谢您的帮助。

编辑：

我的代码：

conn = psycopg2.extras.LoggingConnection(DSN)
File=open('log.log','a')
File.write('================================')
psycopg2.extras.LoggingConnection.initialize(conn,File)

File.write('\n'+time.strftime("%Y-%m-%d %H:%M:%S") + '---Executing query:\n\t')
q="""insert into city(id, name, countrycode, district, population) values (4080,'Sevilla', 'ESP', 'andalucia', 10000)"""
c=conn.cursor()
c.execute(q)
File.write('\n'+time.strftime("%Y-%m-%d %H:%M:%S") + '---Executing query:\n\t')
q="""delete from city where id = 4080"""
c=conn.cursor()
c.execute(q)
conn.commit()
File.close()

这是我的输出日志：

================================
2012-12-30 22:42:31---Executing query:
    insert into city(id, name, countrycode, district, population) values (4080,'Sevilla', 'ESP', 'andalucia', 10000)

2012-12-30 22:42:31---Executing query:
    delete from city where id = 4080

我想在日志文件中查看受影响的行数和错误信息。最后，我想要一个包含所有事件的完整日志文件。

【问题讨论】：

如果您在让示例 LoggingCursor 类工作时遇到问题，请将您的代码连同 SQL 语句的输出一起发布，我们会看看是否可以帮助您。 :)

标签： python postgresql psycopg2 pgadmin

【解决方案1】：

据我所知，LoggingCursor 类没有满足您的三个要求

查询执行时间
受影响的行数
包含所有事件的完整日志文件。

对于第一个要求，请查看 psycopg2.extras 中 MinTimeLoggingConnection 类的源代码。它子类 LoggingConnection 并输出超过最小时间的查询的执行时间（注意这需要与 MinTimeLoggingCursor 结合使用）。

第二个要求，游标类的rowcount attribute指定

最后一次执行*() 产生的行数（对于 DQL SELECT 之类的语句）或受影响的（对于 UPDATE 之类的 DML 语句或插入）

因此，应该可以创建您自己的包含此附加功能的 LoggingConnection 和 LoggingCursor 类型。

我的尝试如下。只需在您的代码中将LoggingConnection 替换为LoggingConnection2，这一切都应该有效。作为旁注，您不需要为第二个查询创建新游标。定义第二个查询后，您可以再次调用 c.execute(q)。

import psycopg2
import os
import time
from psycopg2.extras import LoggingConnection
from psycopg2.extras import LoggingCursor

class LoggingConnection2(psycopg2.extras.LoggingConnection):
    def initialize(self, logobj):
        LoggingConnection.initialize(self, logobj)

    def filter(self, msg, curs):
        t = (time.time() - curs.timestamp) * 1000
        return msg + os.linesep + 'Query executed in: {0:.2f} ms. {1} row(s) affected.'.format(t, curs.rowcount)

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

class LoggingCursor2(psycopg2.extras.LoggingCursor):    
    def execute(self, query, vars=None):
        self.timestamp = time.time()
        return LoggingCursor.execute(self, query, vars)

    def callproc(self, procname, vars=None):
        self.timestamp = time.time()
        return LoggingCursor.execute(self, procname, vars)

我不确定如何创建所有事件的完整日志，但连接类的 notices attribute 可能会很有趣。

【讨论】：

为我工作:)

【解决方案2】：

也许您无需编写任何代码即可获得所需的内容。

postgresql 本身有一个名为“log_min_duration”的选项可能会对您有所帮助。

您可以将其设置为零，并且将记录每个查询及其运行时成本。或者您可以将其设置为某个正数，例如 500，而 postgresql 只会记录运行时间至少为 500 毫秒的查询。

您不会在日志文件中获得查询结果，但您会获得准确的查询，包括内插的绑定参数。

如果这对您很有效，请稍后查看 auto_explain 模块。

祝你好运！

【讨论】：

【解决方案3】：

看看 LoggingCursor 是如何实现的，编写自己的游标子类：非常简单。

【讨论】：