当然可以用Django ORM实现这个需求，不需要非得依赖原生SQL！不过两种方式各有适用场景，我给你详细拆解一下：

一、使用Django ORM实现 #

ORM的优势是可读性强、符合Django的开发习惯，而且能自动处理数据库兼容性问题。这里的核心是利用ContentType和预取查询（Prefetch）来避免N+1查询，同时整理出你需要的结构。

步骤1：导入依赖并定义ContentType #

首先我们需要获取目标模型的ContentType，用来筛选对应的贡献记录：

from django.contrib.contenttypes.models import ContentType
from django.db.models import Prefetch
from .models import User, UserContribution, Tip, Example, Struggle

# 获取各模型对应的ContentType
tip_ct = ContentType.objects.get_for_model(Tip)
example_ct = ContentType.objects.get_for_model(Example)
struggle_ct = ContentType.objects.get_for_model(Struggle)

步骤2：创建预取规则 #

用Prefetch对象分别预取每个用户的三类贡献，这样可以批量查询，避免多次数据库请求：

# 预取用户的Tip贡献
prefetch_tips = Prefetch(
    'usercontribution_set',
    queryset=UserContribution.objects.filter(content_type=tip_ct),
    to_attr='tip_contributions'
)

# 预取用户的Example贡献
prefetch_examples = Prefetch(
    'usercontribution_set',
    queryset=UserContribution.objects.filter(content_type=example_ct),
    to_attr='example_contributions'
)

# 预取用户的Struggle贡献
prefetch_struggles = Prefetch(
    'usercontribution_set',
    queryset=UserContribution.objects.filter(content_type=struggle_ct),
    to_attr='struggle_contributions'
)

步骤3：获取用户并整理数据 #

先筛选目标用户，然后预取所有贡献，最后转换成你需要的JSON结构。这里我们用批量查询来避免N+1问题：

# 选择目标用户（这里示例选择id为1、2的用户）
selected_users = User.objects.filter(id__in=[1, 2]).prefetch_related(
    prefetch_tips,
    prefetch_examples,
    prefetch_struggles
)

# 批量获取所有需要的模型实例，减少查询次数
# 收集所有Tip的ID
all_tip_ids = []
for user in selected_users:
    all_tip_ids.extend([contrib.object_id for contrib in user.tip_contributions])
tip_map = {tip.id: tip for tip in Tip.objects.filter(id__in=all_tip_ids)}

# 收集所有Example的ID
all_example_ids = []
for user in selected_users:
    all_example_ids.extend([contrib.object_id for contrib in user.example_contributions])
example_map = {ex.id: ex for ex in Example.objects.filter(id__in=all_example_ids)}

# 收集所有Struggle的ID
all_struggle_ids = []
for user in selected_users:
    all_struggle_ids.extend([contrib.object_id for contrib in user.struggle_contributions])
struggle_map = {struggle.id: struggle for struggle in Struggle.objects.filter(id__in=all_struggle_ids)}

# 整理成目标格式
result = []
for user in selected_users:
    user_data = {
        'user_id': user.id,
        'first_name': user.first_name,
        'last_name': user.last_name,
        'tips': [],
        'examples': [],
        'struggles': []
    }

    # 填充Tip数据
    for contrib in user.tip_contributions:
        tip = tip_map[contrib.object_id]
        user_data['tips'].append({
            'id': tip.id,
            'title': tip.title,
            'content': tip.content,
            'contributed_at': contrib.contributed_at.strftime('%Y-%m-%d')
        })

    # 填充Example数据
    for contrib in user.example_contributions:
        example = example_map[contrib.object_id]
        user_data['examples'].append({
            'id': example.id,
            'headline': example.headline,
            'content': example.content,
            'contributed_at': contrib.contributed_at.strftime('%Y-%m-%d')
        })

    # 填充Struggle数据
    for contrib in user.struggle_contributions:
        struggle = struggle_map[contrib.object_id]
        user_data['struggles'].append({
            'id': struggle.id,
            'headline': struggle.headline,
            'content': struggle.content,
            'contributed_at': contrib.contributed_at.strftime('%Y-%m-%d')
        })

    result.append(user_data)

二、原生SQL的高效实现方式 #

如果你的用户量和贡献量很大，原生SQL会更高效——它可以通过一次查询完成所有关联和聚合，避免ORM中多次查询和内存拼接的开销。这里我们用PostgreSQL的JSON函数来直接生成你需要的结构（如果用其他数据库，语法会略有不同，但思路一致）。

原生SQL查询语句 #

SELECT
    u.id AS user_id,
    u.first_name,
    u.last_name,
    -- 聚合Tip数据，过滤掉非Tip的记录
    JSON_AGG(
        JSON_BUILD_OBJECT(
            'id', t.id,
            'title', t.title,
            'content', t.content,
            'contributed_at', uc.contributed_at::DATE
        )
    ) FILTER (WHERE ct.model = 'tip') AS tips,
    -- 聚合Example数据
    JSON_AGG(
        JSON_BUILD_OBJECT(
            'id', e.id,
            'headline', e.headline,
            'content', e.content,
            'contributed_at', uc.contributed_at::DATE
        )
    ) FILTER (WHERE ct.model = 'example') AS examples,
    -- 聚合Struggle数据
    JSON_AGG(
        JSON_BUILD_OBJECT(
            'id', s.id,
            'headline', s.headline,
            'content', s.content,
            'contributed_at', uc.contributed_at::DATE
        )
    ) FILTER (WHERE ct.model = 'struggle') AS struggles
FROM
    auth_user u
JOIN
    user_contribution uc ON u.id = uc.contributed_by_id
JOIN
    django_content_type ct ON uc.content_type_id = ct.id
LEFT JOIN
    tip t ON ct.id = (SELECT id FROM django_content_type WHERE model = 'tip') AND uc.object_id = t.id
LEFT JOIN
    example e ON ct.id = (SELECT id FROM django_content_type WHERE model = 'example') AND uc.object_id = e.id
LEFT JOIN
    struggle s ON ct.id = (SELECT id FROM django_content_type WHERE model = 'struggle') AND uc.object_id = s.id
WHERE
    u.id IN %s
GROUP BY
    u.id, u.first_name, u.last_name;

在Django中执行原生SQL #

from django.db import connection

def get_user_contributions(selected_user_ids):
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                u.id AS user_id,
                u.first_name,
                u.last_name,
                JSON_AGG(
                    JSON_BUILD_OBJECT(
                        'id', t.id,
                        'title', t.title,
                        'content', t.content,
                        'contributed_at', uc.contributed_at::DATE
                    )
                ) FILTER (WHERE ct.model = 'tip') AS tips,
                JSON_AGG(
                    JSON_BUILD_OBJECT(
                        'id', e.id,
                        'headline', e.headline,
                        'content', e.content,
                        'contributed_at', uc.contributed_at::DATE
                    )
                ) FILTER (WHERE ct.model = 'example') AS examples,
                JSON_AGG(
                    JSON_BUILD_OBJECT(
                        'id', s.id,
                        'headline', s.headline,
                        'content', s.content,
                        'contributed_at', uc.contributed_at::DATE
                    )
                ) FILTER (WHERE ct.model = 'struggle') AS struggles
            FROM
                auth_user u
            JOIN
                user_contribution uc ON u.id = uc.contributed_by_id
            JOIN
                django_content_type ct ON uc.content_type_id = ct.id
            LEFT JOIN
                tip t ON ct.id = (SELECT id FROM django_content_type WHERE model = 'tip') AND uc.object_id = t.id
            LEFT JOIN
                example e ON ct.id = (SELECT id FROM django_content_type WHERE model = 'example') AND uc.object_id = e.id
            LEFT JOIN
                struggle s ON ct.id = (SELECT id FROM django_content_type WHERE model = 'struggle') AND uc.object_id = s.id
            WHERE
                u.id IN %s
            GROUP BY
                u.id, u.first_name, u.last_name;
        """, (tuple(selected_user_ids),))
        rows = cursor.fetchall()
        
        # 转换为目标格式
        result = []
        for row in rows:
            user_id, first_name, last_name, tips, examples, struggles = row
            result.append({
                'user_id': user_id,
                'first_name': first_name,
                'last_name': last_name,
                'tips': tips if tips else [],
                'examples': examples if examples else [],
                'struggles': struggles if struggles else []
            })
        return result

总结 #

• ORM方式：适合中小数据量，代码易维护、可读性强，符合Django生态。
• 原生SQL方式：适合大数据量场景，查询效率更高，直接在数据库层面完成聚合和格式转换。

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