Django自定义命令无法创建新条目问题求助

问题描述 #

作为Django新手，开发了一个每日为多个地点生成JSON预报数据的应用，通过startjobs.py配置定时任务运行生成函数。任务显示成功运行，但未新增数据库Forecast条目，也未创建目标JSON文件，需要排查修复。

现有代码 #

模型代码 #

class Location(models.Model):
    location = models.CharField(max_length = 75, unique = True)

    def __str__(self):
        return f"{self.location}"


INTERVAL_CHOICES = [
    ('today', 'Today'),
    ('hourly', 'Hourly'),
    ('daily', 'Daily'),
]

class Forecast(models.Model):
    location =  models.ForeignKey(Location, on_delete = models.CASCADE)
    date = models.DateField()
    interval = models.CharField(max_length = 6, choices = INTERVAL_CHOICES, default = 'today')
    filename = models.FileField(upload_to = 'test/')

    def __str__(self):
        return f"Forecast for max wave height - {self.interval} in {self.location} on {self.date}"

startjobs.py代码 #

import datetime as dt
from django.core.management.base import BaseCommand
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.jobstores.django import DjangoJobStore
from django.conf import settings
import logging
from .models import Location, Forecast
import fg_daily, fg_hourly  # 自定义预报生成模块

logger = logging.getLogger(__name__)

locations = Location.objects.values_list("location", flat=True)

def save_new_forecasts():

    date = dt.date.today().strftime("%Y-%m-%d")

    for location in locations:
        if not Forecast.objects.filter(date = date).exists:
            daily_data = create_daily_forecast(location)
            hourly_data = create_hourly_forecast(location)
            daily_forecast = Forecast(
                location = location,
                date = date,
                interval = daily_data[0],
                filename = daily_data[1]
            )
            daily_forecast.save()
            hourly_forecast = Forecast(
                location_id = location,
                date = date,
                interval = hourly_data[0],
                filename = hourly_data[1]
            )
            hourly_forecast.save()

def create_daily_forecast(location):

    interval = "daily"
    filename = fg_daily.generate_daily_forecast(location, 10)

    return [interval, filename]

def create_hourly_forecast(location):
   
    location_id = location
    date = dt.date.today().strftime("%Y.%m.%d")
    interval = "hourly"
    filename = fg_hourly.generate_hourly_forecast(location, 24)

    return [location_id, date, interval, filename]

def delete_old_job_executions(max_age=604_800):
    from apscheduler.executors.django import DjangoJobExecution
    DjangoJobExecution.objects.delete_old_job_executions(max_age)

class Command(BaseCommand):
    help = "Runs apscheduler."

    def handle(self, *args, **options):
        scheduler = BlockingScheduler(timezone=settings.TIME_ZONE)
        scheduler.add_jobstore(DjangoJobStore(), "default")

        scheduler.add_job(
            save_new_forecasts,
            trigger="interval",
            minutes=2,
            id="Forecasts",
            max_instances=1,
            replace_existing=True,
        )
        logger.info("Added job: Save New Forecasts")

        scheduler.add_job(
            delete_old_job_executions,
            trigger=CronTrigger(
                day_of_week="mon", hour="00", minute="00"
            ),  # 每周一午夜执行
            id="Delete Old Job Executions",
            max_instances=1,
            replace_existing=True,
        )
        logger.info("Added weekly job: Delete Old Job Executions.")

        try:
            logger.info("Starting scheduler...")
            scheduler.start()
        except KeyboardInterrupt:
            logger.info("Stopping scheduler...")
            scheduler.shutdown()
            logger.info("Scheduler shut down successfully!")

JSON生成辅助函数 #

import pandas as pd
import os

def generate_json(location, interval, data):
    df = pd.DataFrame(data=data)
    filename = generate_filename(location, interval)
    df.to_json(f'{PATH}/{filename}', orient='records')
    return filename

def generate_daily_forecast(location, length):
    date_list = generate_date_list(length)
    height_list = generate_height_list(length)
    keys = ["date", "max_wave_height"]
    values = ['', '']
    data = []

    for dates in date_list:
        i = date_list.index(dates)
        values[0] = date_list[i]
        values[1] = height_list[i]
        pre_data = dict(zip(keys, values))
        data.append(pre_data)

    filename = generate_json(location, 'daily', data)
    return filename

核心问题与修复方案 #

1. exists方法未调用 #

原代码中if not Forecast.objects.filter(date = date).exists未加括号，exists是QuerySet的方法，必须调用exists()才会执行数据库查询，否则只是引用方法对象，判断永远为False，导致后续逻辑完全不执行。

修复：

if not Forecast.objects.filter(location=location_obj, date=today).exists():

2. 外键关联错误 #

• locations = Location.objects.values_list("location", flat=True)获取的是location字段的字符串值，而Forecast的location字段是ForeignKey，需要传入Location对象或数据库id（整数），直接传字符串会导致外键关联失败。
• location_id = location同样错误，字符串无法匹配数据库id。

修复：
获取完整的Location对象，而不是仅字段值：

# 替换原locations定义
locations = Location.objects.all()

# 循环时使用对象
for location_obj in locations:
    # 创建Forecast时传入对象
    daily_forecast = Forecast(
        location=location_obj,
        ...
    )

3. 返回值对应错误 #

create_hourly_forecast返回[location_id, date, interval, filename]，但调用时错误地用hourly_data[0]作为interval，hourly_data[1]作为filename，完全不匹配。

修复：
调整返回值结构，和create_daily_forecast保持一致：

def create_hourly_forecast(location):
    interval = "hourly"
    filename = fg_hourly.generate_hourly_forecast(location, 24)
    return [interval, filename]

# 调用时对应正确的索引
hourly_data = create_hourly_forecast(location_obj.location)
hourly_forecast = Forecast(
    location=location_obj,
    date=today,
    interval=hourly_data[0],
    filename=hourly_data[1]
)

4. 日期类型不匹配 #

原代码中date = dt.date.today().strftime("%Y-%m-%d")将日期转为字符串，但Forecast的date字段是DateField，需要传入datetime.date对象，直接传字符串会导致字段类型错误。

修复：

today = dt.date.today()  # 直接使用date对象

5. FileField路径处理错误 #

直接传入文件名字符串给FileField可能无法正确关联到存储路径，需要确保文件生成在MEDIA_ROOT下，并且返回相对路径。

修复：

# 在generate_json中使用MEDIA_ROOT路径
from django.conf import settings

def generate_json(location, interval, data):
    df = pd.DataFrame(data=data)
    filename = generate_filename(location, interval)
    # 确保文件保存在MEDIA_ROOT/test/目录下
    save_dir = os.path.join(settings.MEDIA_ROOT, 'test')
    os.makedirs(save_dir, exist_ok=True)  # 确保目录存在
    save_path = os.path.join(save_dir, filename)
    df.to_json(save_path, orient='records')
    # 返回相对MEDIA_ROOT的路径，对应FileField的upload_to
    return os.path.join('test', filename)

修正后的关键函数示例 #

def save_new_forecasts():
    today = dt.date.today()
    locations = Location.objects.all()

    for location_obj in locations:
        # 按地点和日期过滤，避免重复生成
        if not Forecast.objects.filter(location=location_obj, date=today).exists():
            # 生成每日预报
            daily_interval, daily_filename = create_daily_forecast(location_obj.location)
            daily_forecast = Forecast(
                location=location_obj,
                date=today,
                interval=daily_interval,
                filename=daily_filename
            )
            daily_forecast.save()

            # 生成小时预报
            hourly_interval, hourly_filename = create_hourly_forecast(location_obj.location)
            hourly_forecast = Forecast(
                location=location_obj,
                date=today,
                interval=hourly_interval,
                filename=hourly_filename
            )
            hourly_forecast.save()

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