如何将指定SQL查询转换为Django ORM？子查询报错求解

给定SQL查询语句：

SELECT messages1.id, messages1.channel FROM (SELECT * FROM messages WHERE timestamp >= datetime.now() AND channel IN ('ALL', 'UC')) messages1 LEFT JOIN read_messages ON messages1.id = read_messages.message_id WHERE read_messages.id IS NULL;

对应的Django模型：

from django.db import models


class Messages(models.Model):
    channel = models.TextField(blank=True, null=True)
    message = models.TextField(blank=True, null=True)
    timestamp = models.TextField(blank=True, null=True)
    publisher = models.IntegerField(blank=True, null=True)
    type = models.CharField(max_length=5)

    class Meta:
        managed = False
        db_table = 'messages'


class NotifiedMessages(models.Model):
    id = models.IntegerField(primary_key=True)
    user_id = models.IntegerField()
    message_id = models.IntegerField()

    class Meta:
        managed = False
        db_table = 'notified_messages'


class ReadMessages(models.Model):
    user_id = models.IntegerField(blank=True, null=True)
    message_id = models.IntegerField(blank=True, null=True)

    class Meta:
        managed = False
        db_table = 'read_messages'

尝试转换时出现错误：django.db.utils.ProgrammingError: subquery must return only one column，尝试的代码如下：

ts = Messages.objects.filter(timestamp__gte=datetime.today(), channel__in=['ALL', 'UC'])
Messages.objects.annotate(messages1=Subquery(ts, output_field=models.CharField()))

请问上述代码存在什么问题？如何正确将该SQL查询转换为Django ORM？

你的代码问题 #

你用Subquery时传入的是整个Messages查询集（ts），这个查询集会返回Messages的所有字段，而Subquery要求子查询只能返回单个字段，这就是报错的核心原因。另外你的写法完全偏离了原SQL的逻辑——原SQL是做左连接后筛选未被标记为已读的消息，不是给主查询加子查询注解。

正确的Django ORM写法 #

原SQL的核心逻辑是：

1. 筛选messages表中时间≥当前时间、频道为'ALL'或'UC'的记录
2. 左连接read_messages表，筛选出在read_messages中无对应记录的消息（即未被读取的消息）
3. 返回这些消息的id和channel字段

对应两种实现方式：

方式一：exclude结合子查询 #

先获取所有已读消息的message_id，再从符合条件的Messages中排除这些消息：

from django.db.models import Subquery
import datetime

# 获取所有已读消息的message_id列表
read_msg_ids = ReadMessages.objects.values_list('message_id', flat=True)

# 筛选目标消息并排除已读项，只取id和channel字段
result = Messages.objects.filter(
    timestamp__gte=datetime.datetime.now(),
    channel__in=['ALL', 'UC']
).exclude(
    id__in=read_msg_ids
).values('id', 'channel')

方式二：左连接+isnull筛选 #

直接模拟原SQL的左连接逻辑，利用Django的反向关联自动处理连接：

import datetime

result = Messages.objects.filter(
    timestamp__gte=datetime.datetime.now(),
    channel__in=['ALL', 'UC'],
    readmessages__isnull=True  # 反向关联名称为小写模型名，可通过related_name自定义
).values('id', 'channel')

两种写法都会生成与原SQL逻辑一致的查询，最终返回指定字段的结果集。

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