Python如何延迟1小时执行任务？激活码过期调度遇错求替代库

我正在开发一个需要激活码的用户注册功能，希望激活码1小时后过期，编写了如下代码：

def register(request):
    form = CreateUserForm()
    if request.method == "POST":
        form = CreateUserForm(request.data)
        if form.is_valid():
            email = request.POST.get('email')
            password = request.POST.get('password')
            user = get_user_model()
            user = user_manager.create_user(email=email, password=password)

            # Generate an activation code and store it in the database
            code = generate_activation_code()
            # activation = Activations.objects.create(user=12, code=code)

            # Schedule the activation code for deletion after 1 hour
            schedule_activation_code_deletion(user, code)
            # schedule_activation_code_deletion(user, code)

            # Send the activation code to the user's email address
            subject = "Your verification code"
            message = f"Your verification code is {code}"
            send_email(subject, message, settings.EMAIL_HOST_USER, email)
            return Response('User created.')
    return Response(status=400)

def schedule_activation_code_deletion(user_id, code):
    queue = django_rq.get_queue('default')
    queue.enqueue_in(datetime.timedelta(minutes=1), test)

运行代码时遇到错误：

Error while reading from localhost:5432 : (10054, 'An existing connection was forcibly closed by the remote host', None, 10054, None)

请问有没有其他可用的库可以实现激活码1小时后过期的功能？

以下几种方案可以实现你的需求，避开当前的数据库连接问题：

1. 无需定时任务：存储过期时间，验证时判断 #

这是最简洁的方案，不需要依赖任何任务队列库。创建激活码时直接记录当前时间+1小时的过期时间，用户验证激活码时先检查是否在有效期内：

# 创建激活码时
activation = Activations.objects.create(
    user=user,
    code=code,
    expires_at=datetime.datetime.now() + datetime.timedelta(hours=1)
)

# 验证激活码的逻辑
def verify_code(request):
    code = request.POST.get('code')
    try:
        # 只查询未过期的激活码
        activation = Activations.objects.get(code=code, expires_at__gt=datetime.datetime.now())
        # 激活用户
        activation.user.is_active = True
        activation.user.save()
        # 验证后删除激活码
        activation.delete()
        return Response('激活成功。')
    except Activations.DoesNotExist:
        return Response('激活码无效或已过期。', status=400)

这种方案完全避免了定时任务带来的数据库连接问题，适合大多数场景。

2. 使用Celery实现定时删除 #

Celery是Python生态中成熟的异步任务队列，支持定时任务。配置完成后，可实现1小时后自动删除激活码：

# 定义异步任务（通常放在tasks.py中）
from celery import shared_task

@shared_task
def delete_activation_code(user_id, code):
    Activations.objects.filter(user=user_id, code=code).delete()

# 在注册函数中调用定时任务
def schedule_activation_code_deletion(user_id, code):
    # countdown参数设置为3600秒（1小时）
    delete_activation_code.apply_async(args=[user_id, code], countdown=3600)

3. 使用Django Q #

Django Q是轻量级的Django专属任务队列，配置简单，支持定时任务：

# 安装并配置好Django Q后，编写定时逻辑
from django_q.tasks import async_task

def schedule_activation_code_deletion(user_id, code):
    async_task(
        'your_app.tasks.delete_activation_code',  # 指向任务函数的路径
        user_id,
        code,
        schedule_type='O',  # 一次性任务
        next_run=datetime.datetime.now() + datetime.timedelta(hours=1)
    )

# 对应的删除任务（your_app/tasks.py）
def delete_activation_code(user_id, code):
    Activations.objects.filter(user=user_id, code=code).delete()

4. 使用APScheduler #

APScheduler是通用的Python定时任务库，可轻松集成到Django项目中：

# 初始化调度器（可放在项目的apps.py或单独的配置文件中）
from apscheduler.schedulers.background import BackgroundScheduler
import django

django.setup()
scheduler = BackgroundScheduler()
scheduler.start()

# 定义删除任务
def delete_activation_code(user_id, code):
    Activations.objects.filter(user=user_id, code=code).delete()

# 调度任务
def schedule_activation_code_deletion(user_id, code):
    scheduler.add_job(
        delete_activation_code,
        'date',
        run_date=datetime.datetime.now() + datetime.timedelta(hours=1),
        args=[user_id, code]
    )

注意生产环境中需要确保调度器随Django进程持续运行，可配合supervisor等进程管理工具。

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