Django中创建模型时生成多模型实例的实现方案求助

我想实现的功能：用户在表单输入关键词提交搜索，搜索请求会保存到TweetSearch模型。之后运行get_tweets()函数抓取含该关键词的推文，保存到TweetInstance模型，最后在仪表盘中展示TweetInstance的部分数据。

我不清楚怎么在单个视图里创建多个模型实例，所以在搜索表单和仪表盘之间加了一个视图来运行get_tweets()并存数据，但效果不好，肯定有更优的实现方式。

另一种思路是把这个函数放进dashboard()视图，但这样每次POST请求都会执行该函数，导致网站运行极慢。

求帮助解决这个问题。

models.py #

from datetime import datetime, timedelta
import uuid
from django.db import models
from django.contrib.auth.models import User

class TweetSearch(models.Model):
    search_term = models.CharField(max_length=200, blank=True)
    QUERY_CHOICES = (
        ('t', 'in tweet'),
        ('h', 'in hashtag'),
        ('u', 'username'),
    )
    id = models.UUIDField(primary_key=True, default=uuid.uuid4)
    query_type = models.CharField(max_length=1, choices=QUERY_CHOICES, blank=True, default='t')
    start_default = datetime.now() - timedelta(days=30)
    start_date = models.DateTimeField(default=start_default)
    end_date = models.DateTimeField(default=datetime.now)
    language = models.CharField(max_length=200, blank=True, null=True)
    country = models.CharField(max_length=200, blank=True, null=True)
    city = models.CharField(max_length=200, blank=True, null=True)
    searcher = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
    created = models.DateTimeField(auto_now_add=True, null=True)

    def __str__(self):
        return f"Tweets with the word {self.search_term} from {self.start_date} till {self.end_date} written in " \
               f"{self.language} in {self.country}."

class TweetInstance(models.Model):
    tweet_search = models.ForeignKey('TweetSearch', on_delete=models.RESTRICT, null=True)
    content = models.CharField(max_length=280, blank=True, null=True)
    date = models.DateTimeField(blank=True, null=True)
    url = models.CharField(max_length=500, blank=True, null=True)
    tweet_id = models.CharField(max_length=200, blank=True, null=True)
    username = models.CharField(max_length=200, blank=True, null=True)
    followers_count = models.IntegerField(blank=True, null=True)
    reply_count = models.IntegerField(blank=True, null=True)
    retweet_count = models.IntegerField(blank=True, null=True)
    like_count = models.IntegerField(blank=True, null=True)
    quote_count = models.IntegerField(blank=True, null=True)
    language = models.CharField(max_length=200, blank=True, null=True)
    outlinks = models.CharField(max_length=500, blank=True, null=True)
    media = models.CharField(max_length=500, blank=True, null=True)
    retweeted_tweet = models.CharField(max_length=280, blank=True, null=True)
    quoted_tweet = models.CharField(max_length=280, blank=True, null=True)
    in_reply_to_tweet_id = models.CharField(max_length=200, blank=True, null=True)
    in_reply_to_user = models.CharField(max_length=200, blank=True, null=True)
    mentioned_users = models.CharField(max_length=200, blank=True, null=True)
    long = models.FloatField(null=True, blank=True)
    lat = models.FloatField(null=True, blank=True)
    country = models.CharField(max_length=200, blank=True, null=True)
    city = models.CharField(max_length=200, blank=True, null=True)
    hashtags = models.CharField(max_length=200, blank=True, null=True)
    chashtags = models.CharField(max_length=200, blank=True, null=True)

views.py #

from django.views.generic.edit import CreateView
from django.urls import reverse_lazy
from django.shortcuts import render
import pandas as pd
from .models import TweetSearch, TweetInstance
from .utils import get_tweets  # 假设get_tweets在utils.py中

class TweetCreate(CreateView):
    model = TweetSearch
    form_class = TweetForm
    success_url = reverse_lazy('create_dashboard')

def create_dashboard(request):
    qs = TweetSearch.objects.all()
    search_data = [
        {
            'search_term': x.search_term,
            'created': x.created
        } for x in qs
    ]
    df = pd.DataFrame(search_data)
    df = df.iloc[-1]
    tweets_df = get_tweets(df['search_term'])
    mylist = []
    tweet_search = TweetSearch.objects.latest('created')

    for i in range(0, len(tweets_df['content'])):
        mylist.append(TweetInstance(
            tweet_search=tweet_search,
            date=tweets_df['date'].iloc[i],
            content=tweets_df['content'].iloc[i],
            url=tweets_df['url'].iloc[i],
            username=tweets_df['username'].iloc[i],
            tweet_id=tweets_df['id'].iloc[i],
            followers_count=tweets_df['followers_count'].iloc[i],
            reply_count=tweets_df['reply_count'].iloc[i],
            retweet_count=tweets_df['retweet_count'].iloc[i],
            like_count=tweets_df['like_count'].iloc[i],
            quote_count=tweets_df['quote_count'].iloc[i],
            language=tweets_df['language'].iloc[i],
            outlinks=tweets_df['outlinks'].iloc[i],
            media=tweets_df['media'].iloc[i],
            retweeted_tweet=tweets_df['retweeted_tweet'].iloc[i],
            quoted_tweet=tweets_df['quoted_tweet'].iloc[i],
            in_reply_to_tweet_id=tweets_df['in_reply_to_tweet_id'].iloc[i],
            in_reply_to_user=tweets_df['in_reply_to_user'].iloc[i],
            mentioned_users=tweets_df['mentioned_users'].iloc[i],
            long=tweets_df['long'].iloc[i],
            lat=tweets_df['lat'].iloc[i],
            country=tweets_df['country'].iloc[i],
            city=tweets_df['city'].iloc[i],
            hashtags=tweets_df['hashtags'].iloc[i],
            chashtags=tweets_df['chashtags'].iloc[i])
        )
    TweetInstance.objects.bulk_create(mylist)
    return render(request, 'twitter_sentiment/loading.html')

def dashboard(request):
    # 处理仪表盘逻辑
    pass

1. 重写CreateView的form_valid方法（同步方式） #

直接在TweetCreate视图内完成TweetSearch实例创建、推文抓取和TweetInstance保存，无需额外视图，避免多次查询最新数据的问题：

class TweetCreate(CreateView):
    model = TweetSearch
    form_class = TweetForm
    success_url = reverse_lazy('dashboard')  # 提交后直接跳转到仪表盘

    def form_valid(self, form):
        # 先保存搜索请求实例
        self.object = form.save()
        # 调用抓取函数，传入搜索实例的所有参数（比如时间范围、语言等）
        tweets_df = get_tweets(
            self.object.search_term,
            start_date=self.object.start_date,
            end_date=self.object.end_date,
            language=self.object.language,
            country=self.object.country
        )
        
        # 批量创建推文实例
        tweet_instances = []
        for _, row in tweets_df.iterrows():
            tweet_instances.append(TweetInstance(
                tweet_search=self.object,
                date=row['date'],
                content=row['content'],
                url=row['url'],
                username=row['username'],
                tweet_id=row['id'],
                followers_count=row['followers_count'],
                reply_count=row['reply_count'],
                retweet_count=row['retweet_count'],
                like_count=row['like_count'],
                quote_count=row['quote_count'],
                language=row['language'],
                outlinks=row['outlinks'],
                media=row['media'],
                retweeted_tweet=row['retweeted_tweet'],
                quoted_tweet=row['quoted_tweet'],
                in_reply_to_tweet_id=row['in_reply_to_tweet_id'],
                in_reply_to_user=row['in_reply_to_user'],
                mentioned_users=row['mentioned_users'],
                long=row['long'],
                lat=row['lat'],
                country=row['country'],
                city=row['city'],
                hashtags=row['hashtags'],
                chashtags=row['chashtags']
            ))
        TweetInstance.objects.bulk_create(tweet_instances)
        return super().form_valid(form)

这种方式逻辑集中，但如果get_tweets()耗时较长，用户会等待页面加载。

2. 异步处理（推荐方案） #

用异步任务队列（如django-q、Celery）把推文抓取逻辑放到后台执行，避免阻塞用户请求：

步骤1：创建异步任务（tasks.py） #

from django.db import transaction
from .models import TweetSearch, TweetInstance
from .utils import get_tweets

def fetch_and_save_tweets(tweet_search_id):
    # 获取对应的搜索实例
    tweet_search = TweetSearch.objects.get(id=tweet_search_id)
    # 抓取推文
    tweets_df = get_tweets(
        tweet_search.search_term,
        start_date=tweet_search.start_date,
        end_date=tweet_search.end_date,
        language=tweet_search.language,
        country=tweet_search.country
    )
    
    # 事务内批量保存，保证数据一致性
    with transaction.atomic():
        tweet_instances = []
        for _, row in tweets_df.iterrows():
            tweet_instances.append(TweetInstance(
                tweet_search=tweet_search,
                date=row['date'],
                content=row['content'],
                # 其他字段...
            ))
        TweetInstance.objects.bulk_create(tweet_instances)

步骤2：修改TweetCreate视图 #

from django.views.generic.edit import CreateView
from django.urls import reverse_lazy
from .models import TweetSearch
from .tasks import fetch_and_save_tweets
from django_q.tasks import async_task  # 若使用django-q

class TweetCreate(CreateView):
    model = TweetSearch
    form_class = TweetForm
    success_url = reverse_lazy('dashboard')

    def form_valid(self, form):
        self.object = form.save()
        # 触发异步任务，后台执行抓取和保存
        async_task('your_app.tasks.fetch_and_save_tweets', self.object.id)
        return super().form_valid(form)

用户提交表单后会立即跳转到仪表盘，推文抓取在后台运行。可在仪表盘添加状态提示，比如“正在抓取推文，请稍后刷新查看”。

3. 仪表盘优化 #

只查询当前用户最新搜索对应的推文，避免全量查询：

def dashboard(request):
    # 获取当前用户最近的搜索记录
    latest_search = TweetSearch.objects.filter(searcher=request.user).latest('created')
    # 获取该搜索下的前20条推文
    tweets = TweetInstance.objects.filter(tweet_search=latest_search)[:20]
    context = {
        'latest_search': latest_search,
        'tweets': tweets
    }
    return render(request, 'twitter_sentiment/dashboard.html', context)

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