Python-Django Rest Framework单查询统计多表总行数问题

无过滤条件的批量统计 #

如果所有表只需要统计总行数，可通过PostgreSQL的UNION ALL将多个COUNT(*)查询合并为单次请求，避免多次数据库往返：

from django.db import connection
from your_app.models import CommonModel

def get_table_row_counts():
    # 获取所有继承自CommonModel的子模型对应的数据库表名
    sub_models = CommonModel.__subclasses__()
    table_names = [model._meta.db_table for model in sub_models]
    
    # 构建每个表的统计子查询
    query_parts = []
    for table in table_names:
        query_parts.append(f"SELECT '{table}' AS table_name, COUNT(*) AS row_count FROM {table}")
    
    # 合并为单查询
    final_sql = " UNION ALL ".join(query_parts)
    
    # 执行并解析结果
    with connection.cursor() as cursor:
        cursor.execute(final_sql)
        results = cursor.fetchall()
    
    # 转换为要求的字典格式
    return {table: count for table, count in results}

带自定义过滤条件的批量统计 #

如果部分表需要添加过滤规则（比如某表只统计状态为激活的数据），只需给对应表的查询添加WHERE子句即可：

from django.db import connection
from your_app.models import CommonModel, Table1, Table2

def get_filtered_table_counts():
    # 定义各表的过滤条件，无过滤的表无需配置
    filter_rules = {
        Table1._meta.db_table: "WHERE status = 1",
        Table2._meta.db_table: "WHERE created_at >= CURRENT_DATE - INTERVAL '30 days'"
    }
    
    sub_models = CommonModel.__subclasses__()
    table_names = [model._meta.db_table for model in sub_models]
    
    query_parts = []
    for table in table_names:
        filter_clause = filter_rules.get(table, "")
        query_parts.append(f"SELECT '{table}' AS table_name, COUNT(*) AS row_count FROM {table} {filter_clause}")
    
    final_sql = " UNION ALL ".join(query_parts)
    
    with connection.cursor() as cursor:
        cursor.execute(final_sql)
        results = cursor.fetchall()
    
    return {table: count for table, count in results}

近似统计（高性能非精确场景） #

如果对统计精度要求不高，可直接查询PostgreSQL的系统视图pg_stat_user_tables获取近似行数，无需全表扫描，性能大幅提升：

from django.db import connection
from your_app.models import CommonModel

def get_approx_table_counts():
    sub_models = CommonModel.__subclasses__()
    table_names = [model._meta.db_table for model in sub_models]
    table_name_list = ", ".join([f"'{t}'" for t in table_names])
    
    sql = f"""
        SELECT relname AS table_name, n_live_tup AS row_count
        FROM pg_stat_user_tables
        WHERE relname IN ({table_name_list})
    """
    
    with connection.cursor() as cursor:
        cursor.execute(sql)
        results = cursor.fetchall()
    
    return {table: count for table, count in results}

注意：n_live_tup来自PostgreSQL的统计信息，更新频率由autovacuum控制，可能存在一定延迟，适合后台统计、仪表盘等非实时精确场景。

集成到DRF视图 #

将上述函数集成到DRF的API视图中即可：

from rest_framework.views import APIView
from rest_framework.response import Response

class TableCountAPIView(APIView):
    def get(self, request):
        counts = get_table_row_counts()  # 或调用过滤版/近似版函数
        return Response(counts)

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