Django 定时删除过期数据：使用 Celery 实现自动清理-Python教程-PHP中文网

Django 定时删除过期数据：使用 Celery 实现自动清理

霞舞

发布： 2025-10-13 12:37:01

原创

443人浏览过

django 定时删除过期数据：使用 celery 实现自动清理

本文介绍了如何在 Django 项目中实现自动删除指定时间段前创建的数据。通过结合 Celery 这一强大的异步任务队列，我们可以设置定时任务，定期清理数据库中不再需要的数据，保持数据整洁，提高系统性能。文章将提供详细的步骤和示例代码，帮助你快速掌握这一实用技巧。

在 Django 项目中，经常需要定期清理过期数据，例如用户会话、临时文件或日志记录。手动执行这些清理操作既繁琐又容易出错。本文将介绍如何使用 Celery 实现定时删除过期数据，以 UserHitCount 模型为例，自动删除 15 天前创建的记录。

1. 安装 Celery 和 Redis (或 RabbitMQ)

Celery 需要一个消息代理来传递任务。常用的消息代理包括 Redis 和 RabbitMQ。这里以 Redis 为例：

pip install celery redis

登录后复制

如果选择 RabbitMQ，则安装 kombu：

pip install celery kombu

登录后复制

2. 配置 Celery

在 Django 项目的根目录下创建一个 celery.py 文件：

# celery.py
import os
from celery import Celery

# 设置 Django 的 settings 模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')  # 将 your_project 替换为你的项目名称

app = Celery('your_project') # 将 your_project 替换为你的项目名称

# 使用 Django settings 文件作为 Celery 的配置源
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自动发现 tasks.py 文件
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

登录后复制

在 your_project/settings.py 文件中添加 Celery 的配置：

# settings.py

CELERY_BROKER_URL = 'redis://localhost:6379/0'  # Redis 作为消息代理
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # 存储任务结果 (可选)
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai' # 设置时区

# 可选：设置定时任务的配置，稍后会用到
CELERY_BEAT_SCHEDULE = {
    'delete_old_user_hit_counts': {
        'task': 'smart_search.tasks.delete_old_user_hit_counts',
        'schedule': crontab(minute=0, hour=0),  # 每天凌晨 0 点执行
    },
}

登录后复制

在 your_project/__init__.py 文件中添加以下代码，确保 Celery 应用在 Django 启动时加载：

# __init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

登录后复制

3. 创建 Celery 任务

在 smart_search 应用目录下创建一个 tasks.py 文件：

怪兽AI数字人

数字人短视频创作，数字人直播，实时驱动数字人

查看详情

# smart_search/tasks.py
from celery import shared_task
from django.utils import timezone
from smart_search.models import UserHitCount
from celery.schedules import crontab

@shared_task
def delete_old_user_hit_counts():
    """
    定时删除 15 天前创建的 UserHitCount 记录。
    """
    cutoff_date = timezone.now() - timezone.timedelta(days=15)
    UserHitCount.objects.filter(created_at__lte=cutoff_date).delete()
    print(f"Deleted UserHitCount records created before {cutoff_date}")

登录后复制

4. 模型定义

确认你的 smart_search/models.py 中包含 UserHitCount 模型，并且 created_at 字段是 DateTimeField 类型：

# smart_search/models.py
from django.db import models
from user_application.models import Registered_user

class UserHitCount(models.Model):
    user = models.OneToOneField(Registered_user, on_delete=models.CASCADE)
    search_count = models.IntegerField(default=0)
    question_count = models.IntegerField(default=0)
    created_at = models.DateTimeField(auto_now_add=True, null=True)

    def __str__(self):
        return f"UserHitCount for {self.user.username}" # 假设 Registered_user 模型有 username 字段

登录后复制

5. 运行 Celery

启动 Redis (如果使用 Redis 作为消息代理):

redis-server

登录后复制

启动 Celery worker：

celery -A your_project worker -l info  # 将 your_project 替换为你的项目名称

登录后复制

启动 Celery beat (用于调度定时任务):

celery -A your_project beat -l info  # 将 your_project 替换为你的项目名称

登录后复制

或者，可以将 Celery beat 作为服务运行，例如使用 celery beat -A your_project -f celerybeat.log -l info，然后使用 nohup 等工具将其放到后台运行。

6. 测试

为了测试，你可以手动调用 delete_old_user_hit_counts 任务：

# 在 Django shell 中
from smart_search.tasks import delete_old_user_hit_counts
delete_old_user_hit_counts.delay()

登录后复制

注意事项和总结

时区配置： 确保 settings.py 中的 CELERY_TIMEZONE 设置正确，避免定时任务在错误的时间执行。
错误处理： 在 delete_old_user_hit_counts 任务中添加适当的错误处理机制，例如使用 try...except 块捕获异常，并将错误信息记录到日志中。
性能优化： 如果数据量很大，可以考虑使用批量删除操作，例如 UserHitCount.objects.filter(...).delete()。此外，可以根据实际情况调整 Celery worker 的并发数，以提高任务处理速度。
监控： 建议使用 Celery Flower 等工具监控 Celery 任务的执行情况，及时发现和解决问题。
替代方案: 虽然 Celery 是一个强大的解决方案，但对于简单的定时任务，也可以考虑使用 Django 的 manage.py 命令结合操作系统的定时任务工具 (如 cron)。