如何从PostgreSQL函数实时获取通知至Python控制台？

问题根源 #

PostgreSQL默认会缓存函数内RAISE INFO的输出，直到函数执行完毕才一次性返回给客户端，所以用conn.notices()只能在函数结束后拿到所有消息。即使用pg_notify，如果客户端没有实时监听通知频道，也会等函数执行完才批量获取。

可行解决方案 #

1. 用Psycopg3异步连接+通知回调 #

这是最简洁的实时处理方式，利用Psycopg3的异步API监听通知频道，函数执行过程中发送的pg_notify消息会被实时捕获。

步骤1：修改PostgreSQL函数

把原有的RAISE INFO替换为pg_notify，指定一个调试用的频道：

CREATE OR REPLACE FUNCTION debug_func() RETURNS void AS $$
BEGIN
    -- 发送实时通知
    PERFORM pg_notify('debug_channel', '开始执行第一步');
    -- 模拟业务逻辑延迟
    PERFORM pg_sleep(1);
    
    PERFORM pg_notify('debug_channel', '完成第一步，启动第二步');
    PERFORM pg_sleep(1);
    
    PERFORM pg_notify('debug_channel', '所有步骤执行完成');
END;
$$ LANGUAGE plpgsql;

步骤2：Python异步监听代码

import asyncio
import psycopg

async def listen_notifications(conn):
    # 订阅指定频道
    await conn.execute("LISTEN debug_channel;")
    # 实时接收通知
    async for notify in conn.notifies():
        print(f"[实时调试] {notify.payload}")
        # 收到结束信号时停止监听
        if notify.payload == '所有步骤执行完成':
            break

async def main():
    # 建立异步连接
    async with await psycopg.AsyncConnection.connect(
        "dbname=你的数据库名 user=你的用户名 password=你的密码 host=localhost"
    ) as conn:
        # 启动通知监听任务
        notify_task = asyncio.create_task(listen_notifications(conn))
        # 执行目标函数
        await conn.execute("SELECT debug_func();")
        # 等待监听任务完成
        await notify_task

if __name__ == "__main__":
    asyncio.run(main())

2. Psycopg2同步模式下的线程+轮询方案 #

如果项目依赖Psycopg2（同步版本），可以用线程执行函数，主线程轮询连接获取通知：

import psycopg2
import time
import threading

# 建立连接，必须设置自动提交
conn = psycopg2.connect(
    "dbname=你的数据库名 user=你的用户名 password=你的密码 host=localhost"
)
conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

def run_debug_func():
    """在子线程中执行函数"""
    cur = conn.cursor()
    cur.execute("SELECT debug_func();")
    cur.close()

# 订阅通知频道
cur = conn.cursor()
cur.execute("LISTEN debug_channel;")
cur.close()

# 启动函数执行线程
threading.Thread(target=run_debug_func).start()

# 主线程轮询获取实时通知
while True:
    conn.poll()
    # 处理所有待接收的通知
    while conn.notifies:
        notify = conn.notifies.pop(0)
        print(f"[实时调试] {notify.payload}")
        if notify.payload == '所有步骤执行完成':
            conn.close()
            exit()
    time.sleep(0.1)

3. 拆分函数为分步调用（适合简单场景） #

如果函数逻辑可以拆分，把大函数拆成多个小函数，每次调用一个小函数后立即获取并打印conn.notices()的内容，实现伪实时输出：

import psycopg2

conn = psycopg2.connect("dbname=你的数据库名 user=你的用户名")
cur = conn.cursor()

# 调用第一步函数
cur.execute("SELECT debug_step1();")
# 打印当前通知
for notice in conn.notices:
    print(notice.strip())
# 清空通知列表
conn.notices.clear()

# 调用第二步函数
cur.execute("SELECT debug_step2();")
for notice in conn.notices:
    print(notice.strip())
conn.notices.clear()

cur.close()
conn.close()

内容的提问来源于stack exchange，提问作者Rush