Django执行原生SQL报Raw query must include the primary key错误求助

问题原因 #

Django的raw()方法有强制要求：当你通过模型类调用raw()执行原生SQL时，查询返回的结果集中必须包含该模型对应的主键字段，Django需要这个字段完成模型实例的映射。
你当前使用TransactionEntries.objects.raw()，对应home_transactionentries表（别名h_te），但你的SELECT语句里没有包含该表的主键h_te.id，因此触发报错。
另外你直接用f-string拼接用户ID到SQL语句里，存在SQL注入风险，不推荐该写法。

解决办法 #

方案1：调整SQL适配raw()方法 #

如果你需要将结果映射为TransactionEntries实例，可在SELECT中补充主键字段，同时改用Django raw() 方法自带的参数化传参能力规避注入风险：

def get(self, request):
    query = """
SELECT  h_te.id,
        u_c.name,
        sum(CASE
            WHEN h_te.to_pay_to_customer > h_te.pay_from_customer THEN
                h_te.to_pay_to_customer - h_te.pay_from_customer
            WHEN h_te.pay_from_customer > h_te.to_pay_to_customer THEN
                h_te.pay_from_customer - h_te.to_pay_to_customer
            ELSE 0
        end) as Total,
        u_c.owner_id
FROM home_transactionentries h_te
INNER JOIN home_transaction h_t
        ON h_te.transaction_id = h_t.id
INNER JOIN users_customer u_c
        ON u_c.id = h_t.customer_id
WHERE u_c.owner_id = %s
GROUP BY h_te.id, u_c.name, u_c.owner_id;
    """
    # 第二个参数传入查询参数，避免SQL注入
    query_set = TransactionEntries.objects.raw(query, [request.user.id])

注意GROUP BY子句中也要同步加上h_te.id，符合SQL聚合分组的语法规则。

方案2：直接用游标执行聚合查询（更推荐） #

你的查询属于聚合统计场景，不需要映射为TransactionEntries模型实例，用数据库游标执行更高效，也不需要调整原有查询逻辑：

from django.db import connection

def get(self, request):
    query = """
SELECT  u_c.name,
        sum(CASE
            WHEN h_te.to_pay_to_customer > h_te.pay_from_customer THEN
                h_te.to_pay_to_customer - h_te.pay_from_customer
            WHEN h_te.pay_from_customer > h_te.to_pay_to_customer THEN
                h_te.pay_from_customer - h_te.to_pay_to_customer
            ELSE 0
        end) as Total,
        u_c.owner_id
FROM home_transactionentries h_te
INNER JOIN home_transaction h_t
        ON h_te.transaction_id = h_t.id
INNER JOIN users_customer u_c
        ON u_c.id = h_t.customer_id
WHERE u_c.owner_id = %s
GROUP BY u_c.name, u_c.owner_id;
    """
    with connection.cursor() as cursor:
        cursor.execute(query, [request.user.id])
        # 直接获取结果元组
        rows = cursor.fetchall()
        # 也可以转成字典格式方便接口返回
        columns = [col[0] for col in cursor.description]
        result = [dict(zip(columns, row)) for row in rows]

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