如何用Python统计PostgreSQL数据库中用户各状态的出现次数

问题说明 #

我在PostgreSQL数据库中有一张表，包含9行数据，字段为username、user1_id和status_id（状态ID）。其中username存在重复值，示例数据如下：

username  user1_id   status_id
--------  --------   ---------
  ali        2          1
  pam        3          1
  pam        3          2
  ali        2          1
  ali        2          2
  pam        3          2
   .         .          .
   .         .          .
  goes-on    goes-on

实际场景中有14个绑定指定user1_id的用户名，需要统计每个用户在不同status_id（状态ID）下的出现次数，期望输出格式：

ali(状态ID=1) = 2
ali(状态ID=2) = 1
pam(状态ID=1) = 1
pam(状态ID=2) = 2

实现方案 #

方案1：SQL前置统计+Python读取输出（推荐） #

利用PostgreSQL的分组统计能力直接计算结果，Python仅负责读取并格式化输出，效率更高。

1. 安装PostgreSQL Python驱动：

pip install psycopg2-binary

2. 编写Python代码：

import psycopg2

# 替换为你的数据库连接信息
db_config = {
    "dbname": "你的数据库名",
    "user": "你的用户名",
    "password": "你的密码",
    "host": "数据库地址",
    "port": "端口号"
}

try:
    # 建立数据库连接
    conn = psycopg2.connect(**db_config)
    cursor = conn.cursor()

    # 执行分组统计SQL
    stats_query = """
        SELECT username, status_id, COUNT(*) AS occurrence
        FROM 你的表名
        GROUP BY username, status_id
        ORDER BY username, status_id;
    """
    cursor.execute(stats_query)
    result_set = cursor.fetchall()

    # 按期望格式输出
    for username, status_id, count in result_set:
        print(f"{username}(状态ID={status_id}) = {count}")

finally:
    # 关闭数据库连接
    if cursor:
        cursor.close()
    if conn:
        conn.close()

方案2：Python读取全量数据后统计 #

如果需要在Python层面处理数据，可先读取所有记录，再用字典统计次数。

1. 同样先安装psycopg2-binary，然后编写代码：

import psycopg2
from collections import defaultdict

db_config = {
    "dbname": "你的数据库名",
    "user": "你的用户名",
    "password": "你的密码",
    "host": "数据库地址",
    "port": "端口号"
}

try:
    conn = psycopg2.connect(**db_config)
    cursor = conn.cursor()

    # 读取所有用户和状态ID数据
    fetch_query = "SELECT username, status_id FROM 你的表名;"
    cursor.execute(fetch_query)
    all_records = cursor.fetchall()

    # 统计每个用户的各状态ID出现次数
    count_map = defaultdict(lambda: defaultdict(int))
    for username, status_id in all_records:
        count_map[username][status_id] += 1

    # 格式化输出结果
    for username in sorted(count_map.keys()):
        for status_id in sorted(count_map[username].keys()):
            print(f"{username}(状态ID={status_id}) = {count_map[username][status_id]}")

finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()

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