如何在Django处理请求前的启动阶段仅执行一次函数？

希望在Django服务器启动时运行一段清理代码，满足以下要求：

• 仅执行一次
• 在处理任何请求前完成
• 执行过程中需要访问数据库

已调研情况：

• 尝试过AppConfig.ready()方案，但官方明确指出该方法内不应访问数据库
• 参考过旧的社区帖子，想了解适配新版本Django的可行方案
• 已找到一种看似可行的实现：

from django.dispatch import receiver
from django.db.backends.signals import connection_created

@receiver(connection_created)
def my_receiver(connection, **kwargs):
    with connection.cursor() as cursor:
        # 执行数据库清理操作
        pass

    connection_created.disconnect(my_receiver)

想知道还有其他思路吗？

除了你提到的connection_created信号方案，以下是几种适配新版本Django的实用思路：

1. 修改WSGI/ASGI入口文件 #

如果用WSGI（wsgi.py）或ASGI（asgi.py）启动服务器，可以在Django应用初始化完成后、启动服务前执行代码。用线程锁或缓存确保多进程环境下只跑一次：

# wsgi.py示例
import os
import threading
from django.core.wsgi import get_wsgi_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')

# 先初始化Django应用
application = get_wsgi_application()

def cleanup_legacy_data():
    from django.db import connection
    with connection.cursor() as cursor:
        # 这里写你的清理逻辑，比如删除遗留数据
        cursor.execute("DELETE FROM your_table WHERE is_legacy = TRUE;")

# 用锁确保单进程内仅执行一次
cleanup_lock = threading.Lock()
if cleanup_lock.acquire(blocking=False):
    try:
        cleanup_legacy_data()
    finally:
        cleanup_lock.release()

2. 自定义管理命令+前置执行 #

创建一个清理数据的自定义管理命令，启动服务器前先执行这个命令：

1. 在你的app下创建management/commands/cleanup_legacy.py：

from django.core.management.base import BaseCommand
from django.db import connection

class Command(BaseCommand):
    help = '清理服务器遗留数据'

    def handle(self, *args, **options):
        with connection.cursor() as cursor:
            # 执行清理逻辑
            cursor.execute("DELETE FROM your_table WHERE is_legacy = TRUE;")
        self.stdout.write(self.style.SUCCESS('遗留数据清理完成'))

2. 启动服务器时先跑命令再启动服务：

python manage.py cleanup_legacy && python manage.py runserver

生产环境用Gunicorn等服务时，同样可以先执行清理命令再启动服务。

3. 使用Django 4.0+的startup信号 #

Django 4.0及以上版本提供了django.startup信号，这个信号在Django初始化完成、准备处理请求时触发，每个进程仅触发一次。可以用它来绑定清理逻辑：

from django.dispatch import receiver
from django.core.signals import startup
from django.db import connection

@receiver(startup)
def cleanup_on_startup(sender, **kwargs):
    with connection.cursor() as cursor:
        # 执行清理逻辑
        cursor.execute("DELETE FROM your_table WHERE is_legacy = TRUE;")
    # 可选：断开接收器，避免后续意外触发
    startup.disconnect(cleanup_on_startup)

4. 缓存标记控制单执行 #

如果你的项目用了缓存（比如Redis、Memcached），可以用缓存标记来确保代码只执行一次，哪怕是多进程环境：

from django.core.cache import cache
from django.db import connection

def cleanup_once():
    # 尝试设置缓存标记，过期时间设为服务器运行周期（比如24小时）
    if cache.add('cleanup_done', '1', 86400):
        with connection.cursor() as cursor:
            # 执行清理逻辑
            cursor.execute("DELETE FROM your_table WHERE is_legacy = TRUE;")

# 在wsgi.py或asgi.py初始化完成后调用
cleanup_once()

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