如何在临时数据库中运行Django计算程序并支持命令行控制？

一、基于Django测试框架改造（最简方案） #

Django自带的TestCase天生支持自动创建临时数据库、测试完成后自动销毁，完全匹配你的核心需求，不需要从零搭建临时库管理逻辑，是最省心的选择：

• 加载dump：在类级别的setUpTestData方法中调用你已实现的dump加载逻辑（比如通过subprocess执行psql导入），这个方法只会运行一次，适合大体积dump的加载。
• 命令行传参：你提到的Stack Overflow方案可行，但用环境变量传参更简单，不需要修改测试runner。直接在测试代码里读取环境变量，运行时通过环境变量传递dump路径等参数即可。
• 执行测量：在测试方法里跳过断言逻辑，直接运行要测量的Django代码，用timeit或其他工具记录耗时、资源占用等数据，直接打印或输出结果。

示例代码：

# myapp/tests.py
import os
import timeit
import subprocess
from django.test import TestCase
from django.conf import settings

class MeasurementTest(TestCase):
    @classmethod
    def setUpTestData(cls):
        # 从环境变量获取dump路径
        dump_path = os.environ.get('DUMP_FILE_PATH')
        if not dump_path:
            raise ValueError("请设置DUMP_FILE_PATH环境变量指定dump文件路径")
        cls._load_dump(dump_path)

    @classmethod
    def _load_dump(cls, dump_path):
        # 调用外部命令导入dump到临时测试库
        db_config = settings.DATABASES['default']
        subprocess.run([
            'psql',
            '-d', db_config['NAME'],
            '-U', db_config['USER'],
            '-f', dump_path
        ], check=True, capture_output=True)

    def test_run_measurement(self):
        # 定义要测量的Django代码逻辑
        def measured_task():
            from myapp.models import MyModel
            # 示例：执行一次复杂查询
            return MyModel.objects.filter(created_at__year=2024).aggregate(total_count=models.Count('id'))
        
        # 执行多次测量取平均值
        run_times = 10
        total_time = timeit.timeit(measured_task, number=run_times)
        avg_time = total_time / run_times
        print(f"执行{run_times}次，平均耗时: {avg_time:.4f} 秒")

运行命令：

DUMP_FILE_PATH=/tmp/test_data_dump.sql python manage.py test myapp.tests.MeasurementTest -v 2

二、自定义manage.py命令（更灵活的替代方案） #

如果觉得测试框架的“测试”属性违和，可以写一个自定义Django管理命令，手动控制临时库的生命周期：

• 直接用Django测试工具包的create_test_db和destroy_test_db方法创建/销毁临时库。
• 利用Django自定义命令的argparse支持，直接接收命令行参数（dump路径、测量次数等），比测试框架传参更直观。
• 完成dump加载后，切换数据库连接到临时库，执行测量逻辑，最后销毁数据库。

示例代码框架：

# myapp/management/commands/run_measurement.py
import argparse
import timeit
import subprocess
from django.core.management.base import BaseCommand
from django.test.utils import create_test_db, destroy_test_db
from django.conf import settings
from django.db import connection

class Command(BaseCommand):
    help = '在临时数据库上运行代码并输出性能测量结果'

    def add_arguments(self, parser):
        parser.add_argument('dump_path', type=str, help='数据库dump文件路径')
        parser.add_argument('--times', type=int, default=10, help='测量执行次数')

    def handle(self, *args, **options):
        dump_path = options['dump_path']
        run_times = options['times']

        # 创建临时测试库
        temp_db_name = create_test_db(verbosity=1)
        try:
            # 导入dump到临时库
            db_config = settings.DATABASES['default']
            subprocess.run([
                'psql',
                '-d', temp_db_name,
                '-U', db_config['USER'],
                '-f', dump_path
            ], check=True)

            # 切换连接到临时库
            connection.settings_dict['NAME'] = temp_db_name
            connection.close()
            connection.connect()

            # 执行测量
            def measured_task():
                from myapp.models import MyModel
                return MyModel.objects.filter(status='active').count()
            
            total_time = timeit.timeit(measured_task, number=run_times)
            avg_time = total_time / run_times
            self.stdout.write(self.style.SUCCESS(f"平均执行时间: {avg_time:.4f} 秒"))

        finally:
            # 确保临时库被销毁
            destroy_test_db(temp_db_name, verbosity=1)

运行命令：

python manage.py run_measurement /tmp/test_dump.sql --times=5

三、关于测试框架传参的可行性 #

你提到的Stack Overflow方案完全可行，只是自定义测试runner的方式略显繁琐。但用环境变量传递参数可以绕过复杂的runner修改，直接在测试代码中读取变量，既利用了测试框架的临时库管理能力，又能灵活传递参数，是性价比最高的实现方式。

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