Django中按时间范围关联多表数据的高效实现方案问询

问题背景 #

我定义了两个Django模型：

class CombinedData(models.Model):
    datetimestamp = models.DateTimeField(null=True, blank=True)
    N1_leq = models.FloatField(null=True, blank=True)
    N2_leq = models.FloatField(null=True, blank=True)


class DateTimeFolderTable(models.Model):
    path = models.CharField(max_length=255,primary_key=True)
    datetimestamp = models.DateTimeField(null=True, blank=True)
    device = models.IntegerField(default=0)

• CombinedData存储等时间间隔的监测数据
• DateTimeFolderTable存储文件路径、随机时间戳及设备编号

需求：为每个CombinedData条目，匹配其所属整小时时间区间内、不同设备对应的DateTimeFolderTable中的文件路径，最终渲染成表格。当前遍历CombinedData逐行过滤的方案效率太低，需要更高效、Pythonic的实现方式。

优化方案 #

核心思路是批量预查询+内存中快速匹配，彻底避免N+1查询问题，利用Django ORM的聚合分组能力减少数据库交互次数，再通过Python字典实现O(1)时间复杂度的匹配。

1. 批量预查询并构建映射字典

先一次性拉取所有需要的数据，再将DateTimeFolderTable的数据按「整小时时间点+设备编号」分组，构建成便于快速查找的字典：

from django.db.models import F, Func, Value
from django.db.models.functions import TruncHour

# 批量获取所有CombinedData（数据量极大时可加分页）
combined_datas = CombinedData.objects.all().order_by('datetimestamp')

# 对DateTimeFolderTable按整小时+设备分组，聚合同组内的文件路径
# MySQL使用GROUP_CONCAT，PostgreSQL替换为StringAgg
folder_groups = DateTimeFolderTable.objects.annotate(
    hour=TruncHour('datetimestamp')  # 提取时间戳的整小时部分
).values('hour', 'device').annotate(
    paths=Func(F('path'), Value(','), function='GROUP_CONCAT')
).values('hour', 'device', 'paths')

# 转换为字典：key=(整小时时间点, 设备编号)，value=路径列表
folder_map = {
    (item['hour'], item['device']): item['paths'].split(',')
    for item in folder_groups
}

PostgreSQL适配代码：

from django.contrib.postgres.aggregates import StringAgg

folder_groups = DateTimeFolderTable.objects.annotate(
    hour=TruncHour('datetimestamp')
).values('hour', 'device').annotate(
    paths=StringAgg('path', delimiter=',')
).values('hour', 'device', 'paths')

2. 内存中匹配每个CombinedData条目

遍历combined_datas，计算每个条目所属的整小时区间，直接从预构建的folder_map中取出对应设备的路径：

result = []
# 提前获取所有存在的设备编号，避免重复遍历字典
all_devices = {device for _, device in folder_map.keys()}

for data in combined_datas:
    if not data.datetimestamp:
        result.append({
            'datetimestamp': data.datetimestamp,
            'N1_leq': data.N1_leq,
            'N2_leq': data.N2_leq,
            'device_paths': {dev: [] for dev in all_devices}
        })
        continue
    
    # 计算当前数据所属的整小时时间点
    data_hour = data.datetimestamp.replace(minute=0, second=0, microsecond=0)
    # 快速匹配各设备的路径
    device_paths = {
        dev: folder_map.get((data_hour, dev), [])
        for dev in all_devices
    }
    
    result.append({
        'datetimestamp': data.datetimestamp,
        'N1_leq': data.N1_leq,
        'N2_leq': data.N2_leq,
        'device_paths': device_paths
    })

3. 渲染表格

拿到result列表后，可直接在Django模板中渲染成目标表格：

{% load custom_tags %} <!-- 加载自定义过滤器 -->
<table border="1">
    <thead>
        <tr>
            <th>时间戳</th>
            <th>N1_leq</th>
            <th>N2_leq</th>
            {% for device in all_devices %}
            <th>设备{{ device }}路径</th>
            {% endfor %}
        </tr>
    </thead>
    <tbody>
        {% for item in result %}
        <tr>
            <td>{{ item.datetimestamp|date:"Y-m-d H:i:s" }}</td>
            <td>{{ item.N1_leq }}</td>
            <td>{{ item.N2_leq }}</td>
            {% for device in all_devices %}
            <td>
                {% for path in item.device_paths|get_item:device %}
                {{ path }}<br>
                {% empty %}
                无匹配路径
                {% endfor %}
            </td>
            {% endfor %}
        </tr>
        {% endfor %}
    </tbody>
</table>

自定义get_item过滤器（用于字典取值），在app/templatetags/custom_tags.py中定义：

from django import template
register = template.Library()

@register.filter
def get_item(dictionary, key):
    return dictionary.get(key, [])

效率对比 #

• 原方案：每个CombinedData条目触发一次数据库查询，总查询次数为1 + N（N为CombinedData条目数）
• 优化后：仅2次数据库查询，后续匹配全在内存中完成，数据量越大，效率提升越显著

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