Python监控PostgreSQL新增数据并实现阈值告警求助

核心修改点 #

• 跟踪最后处理的行标识（假设表有自增主键id，无主键可替换为时间戳等唯一递增字段），确保只处理新增行
• 对新增行的第二、第三列数值做阈值判断，触发对应提示
• 修正原代码中sleep(1)的错误（注释标注每分钟检测，实际应设为sleep(60)）

完整代码 #

import psycopg2
import time

# 数据库连接配置
conn_config = {
    "database": "database",
    "user": "user",
    "password": "password",
    "host": "127.0.0.1",
    "port": "5432"
}

# 初始化最后处理的行ID，初始值0表示从第一行开始
last_processed_id = 0

def check_new_rows():
    global last_processed_id
    conn = None
    cursor = None
    try:
        # 建立数据库连接
        conn = psycopg2.connect(**conn_config)
        conn.autocommit = True
        cursor = conn.cursor()

        # 查询所有未处理的新增行（假设表有自增主键id）
        # 替换column2、column3为你表中第二、第三列的实际列名
        cursor.execute('''SELECT id, column2, column3 FROM today WHERE id > %s ORDER BY id''', (last_processed_id,))
        new_rows = cursor.fetchall()

        for row in new_rows:
            row_id, col2_val, col3_val = row
            # 更新最后处理的ID
            last_processed_id = row_id

            # 第二列数值低于105的判断
            if col2_val < 105:
                print("Warning! The number has dropped below 105.")
            # 第三列数值高于115的判断
            if col3_val > 115:
                print("The number is higher than 115.")

    except psycopg2.Error as e:
        print(f"Database error occurred: {e}")
    finally:
        # 确保关闭游标和连接，避免资源泄漏
        if cursor:
            cursor.close()
        if conn:
            conn.close()

if __name__ == "__main__":
    print("Starting continuous monitoring...")
    while True:
        check_new_rows()
        time.sleep(60)  # 每分钟检测一次新增行

关键说明 #

1. 行跟踪逻辑：通过last_processed_id记录最后处理过的行主键，每次查询只获取大于该ID的行，避免重复处理旧数据
2. 列适配：代码中column2和column3需要替换为你表中第二、第三列的实际列名；如果表无自增主键，可改用创建时间字段（如created_at > last_processed_time）实现新增行跟踪
3. 异常处理：增加数据库操作的异常捕获，避免单次连接失败导致程序崩溃
4. 资源管理：在finally块中强制关闭游标和连接，防止数据库连接泄漏

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