Django 数据库中自动删除过期数据：定时任务实现指南-Python教程-PHP中文网

Django 数据库中自动删除过期数据：定时任务实现指南

碧海醫心

发布： 2025-10-12 08:53:17

原创

322人浏览过

django 数据库中自动删除过期数据：定时任务实现指南

本文旨在指导开发者如何在 Django 项目中实现自动删除数据库中创建时间超过15天的数据。我们将探讨如何利用 Celery 等异步任务队列工具，设置定时任务来定期清理过期数据，并提供示例代码和注意事项，确保数据清理的可靠性和效率。

在 Django 项目中，自动删除数据库中创建时间超过一定期限（例如15天）的数据是一个常见的需求。这种功能有助于保持数据库的整洁，优化性能，并符合数据保留策略。虽然 Django Signals 看起来是一个不错的选择，但在这种场景下，使用定时任务通常是更可靠和高效的方法。本文将介绍如何使用 Celery 配合 Django 实现这一功能。

使用 Celery 实现定时任务

Celery 是一个流行的 Python 异步任务队列，它可以用来执行后台任务，例如定期删除过期数据。以下是使用 Celery 实现自动删除过期数据的步骤：

安装 Celery 和 Redis (或 RabbitMQ):

Celery 需要一个消息代理来传递任务。 Redis 和 RabbitMQ 是常用的选择。
```
pip install celery redis  # 或者 pip install celery rabbitmq
```
登录后复制

配置 Celery:

在你的 Django 项目中，创建一个 celery.py 文件 (通常与 settings.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}')

登录后复制

在 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 任务:

在你的 Django app 中 (例如 smart_search app)，创建一个 tasks.py 文件：

怪兽AI数字人

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

查看详情

# smart_search/tasks.py
from celery import shared_task
from django.utils import timezone
from datetime import timedelta
from .models import UserHitCount

@shared_task
def delete_expired_user_hit_count():
    """
    删除创建时间超过 15 天的 UserHitCount 记录。
    """
    cutoff_date = timezone.now() - timedelta(days=15)
    UserHitCount.objects.filter(created_at__lte=cutoff_date).delete()
    print("Expired UserHitCount records deleted.")

登录后复制

配置 Celery Beat (定时任务调度器):

Celery Beat 用于调度定时任务。在 settings.py 中添加 Celery Beat 的配置：

# settings.py
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'  # 使用数据库存储定时任务
CELERY_BEAT_SCHEDULE = {
    'delete-expired-user-hit-count': {
        'task': 'smart_search.tasks.delete_expired_user_hit_count',  # 任务的完整路径
        'schedule': crontab(minute=0, hour=0),  # 每天凌晨 0 点执行
    },
}

登录后复制

确保你已经安装了 django-celery-beat:

pip install django-celery-beat

登录后复制

并将 django_celery_beat 添加到 INSTALLED_APPS 中：

# settings.py
INSTALLED_APPS = [
    # ...
    'django_celery_beat',
]

登录后复制

运行数据库迁移：

python manage.py migrate

登录后复制

启动 Celery Worker 和 Celery Beat:

打开两个终端窗口，分别运行以下命令：
```
celery -A your_project worker -l info  # 启动 Celery Worker
celery -A your_project beat -l info  # 启动 Celery Beat
```
登录后复制
将 your_project 替换为你的项目名称