Python案例如何批量运行

wen python案例 2026-06-04 421

本文目录导读：

Python案例如何批量运行

使用脚本批量执行
使用exec或importlib动态导入
使用multiprocessing并行运行
使用配置文件管理案例
使用测试框架运行
最实用的完整解决方案
使用建议

我来介绍几种在Python中批量运行案例的方法：

使用脚本批量执行

创建批处理脚本

# batch_run.py
import subprocess
import os
import glob
from pathlib import Path
# 方法1：运行当前目录下所有Python文件
def run_all_python_files():
    python_files = glob.glob("*.py")
    for file in python_files:
        if file != "batch_run.py":  # 跳过自身
            print(f"\n=== 运行 {file} ===")
            subprocess.run(["python", file])
# 方法2：运行指定列表中的文件
def run_specific_files():
    files_to_run = [
        "case1.py",
        "case2.py",
        "case3.py"
    ]
    for file in files_to_run:
        if os.path.exists(file):
            print(f"\n=== 运行 {file} ===")
            result = subprocess.run(["python", file], 
                                  capture_output=True, 
                                  text=True)
            print(result.stdout)
            if result.stderr:
                print(f"错误: {result.stderr}")
# 方法3：带参数的运行
def run_with_args():
    cases = [
        ("case1.py", ["arg1", "arg2"]),
        ("case2.py", ["--input", "data.txt"]),
        ("case3.py", ["--debug"])
    ]
    for file, args in cases:
        print(f"运行 {file} 参数: {args}")
        subprocess.run(["python", file] + args)
if __name__ == "__main__":
    run_all_python_files()

使用`exec`或`importlib`动态导入

# dynamic_run.py
import importlib.util
import sys
from pathlib import Path
def run_module_dynamically(module_path):
    """动态导入并运行模块"""
    module_name = Path(module_path).stem
    # 加载模块
    spec = importlib.util.spec_from_file_location(module_name, module_path)
    module = importlib.util.module_from_spec(spec)
    # 添加模块到sys.modules
    sys.modules[module_name] = module
    # 执行模块
    spec.loader.exec_module(module)
    # 如果模块有main函数，调用它
    if hasattr(module, 'main'):
        module.main()
# 批量运行
cases = ["case1.py", "case2.py", "case3.py"]
for case in cases:
    if Path(case).exists():
        print(f"\n=== 运行 {case} ===")
        try:
            run_module_dynamically(case)
        except Exception as e:
            print(f"运行失败: {e}")

使用`multiprocessing`并行运行

# parallel_run.py
import multiprocessing
import subprocess
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
def run_python_file(file_path):
    """运行单个Python文件"""
    print(f"开始运行: {file_path}")
    result = subprocess.run(["python", file_path], 
                          capture_output=True, 
                          text=True)
    return {
        "file": file_path,
        "success": result.returncode == 0,
        "output": result.stdout,
        "error": result.stderr
    }
def parallel_run():
    """并行运行多个Python文件"""
    files = ["case1.py", "case2.py", "case3.py", "case4.py"]
    # 使用线程池
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(run_python_file, f) for f in files]
        for future in futures:
            result = future.result()
            status = "成功" if result["success"] else "失败"
            print(f"{result['file']}: {status}")
    # 使用进程池（适合CPU密集型任务）
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(run_python_file, files))
        for result in results:
            print(f"{result['file']}: {'成功' if result['success'] else '失败'}")
if __name__ == "__main__":
    parallel_run()

使用配置文件管理案例

# config_run.py
import json
import yaml  # 需要安装: pip install pyyaml
import subprocess
class CaseRunner:
    def __init__(self, config_file):
        self.cases = self.load_config(config_file)
    def load_config(self, config_file):
        """加载配置文件"""
        if config_file.endswith('.json'):
            with open(config_file, 'r') as f:
                return json.load(f)
        elif config_file.endswith('.yaml') or config_file.endswith('.yml'):
            with open(config_file, 'r') as f:
                return yaml.safe_load(f)
        else:
            raise ValueError("不支持的配置文件格式")
    def run_all(self):
        """运行所有案例"""
        results = []
        for case in self.cases.get('cases', []):
            file = case.get('file')
            args = case.get('args', [])
            description = case.get('description', file)
            print(f"\n=== {description} ===")
            try:
                result = subprocess.run(
                    ["python", file] + args,
                    capture_output=True,
                    text=True,
                    timeout=case.get('timeout', 30)  # 超时设置
                )
                results.append({
                    'file': file,
                    'success': result.returncode == 0,
                    'output': result.stdout,
                    'error': result.stderr
                })
                if result.returncode == 0:
                    print("✓ 运行成功")
                else:
                    print(f"✗ 运行失败: {result.stderr}")
            except subprocess.TimeoutExpired:
                print(f"✗ 运行超时")
                results.append({
                    'file': file,
                    'success': False,
                    'error': 'Timeout'
                })
        return results
# 使用示例
runner = CaseRunner('cases_config.json')
results = runner.run_all()

配置文件示例 (cases_config.json):

{
    "cases": [
        {
            "file": "case1.py",
            "description": "基础案例",
            "args": [],
            "timeout": 30
        },
        {
            "file": "case2.py",
            "description": "高级案例",
            "args": ["--input", "data.txt", "--verbose"],
            "timeout": 60
        },
        {
            "file": "test_suite.py",
            "description": "测试套件",
            "args": ["-v"],
            "timeout": 120
        }
    ]
}

使用测试框架运行

# test_runner.py
import unittest
import pytest  # 需要安装: pip install pytest
# 使用unittest
def run_unittest_tests():
    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    # 添加测试类
    suite.addTests(loader.loadTestsFromModule('test_module'))
    suite.addTests(loader.discover('tests', pattern='test_*.py'))
    runner = unittest.TextTestRunner(verbosity=2)
    runner.run(suite)
# 使用pytest
def run_pytest_tests():
    import sys
    # 运行所有测试
    exit_code = pytest.main(['tests/', '-v'])
    # 运行特定标记的测试
    exit_code = pytest.main(['tests/', '-m', 'slow'])
    return exit_code
# 使用pytest参数化运行
import pytest
@pytest.mark.parametrize("case_file", [
    "case1.py",
    "case2.py", 
    "case3.py"
])
def test_case_running(case_file):
    import subprocess
    result = subprocess.run(["python", case_file], 
                          capture_output=True, 
                          text=True)
    assert result.returncode == 0, f"{case_file} 运行失败"

最实用的完整解决方案

# batch_runner.py
import subprocess
import time
from datetime import datetime
from pathlib import Path
import logging
import concurrent.futures
class BatchCaseRunner:
    def __init__(self, cases_dir="cases", log_file="batch_run.log"):
        self.cases_dir = Path(cases_dir)
        self.log_file = log_file
        self.setup_logging()
    def setup_logging(self):
        logging.basicConfig(
            filename=self.log_file,
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
    def run_single_case(self, case_file, timeout=60):
        """运行单个案例"""
        start_time = time.time()
        result = {
            'file': case_file,
            'success': False,
            'duration': 0,
            'output': '',
            'error': ''
        }
        try:
            process = subprocess.run(
                ["python", str(case_file)],
                capture_output=True,
                text=True,
                timeout=timeout
            )
            result['success'] = process.returncode == 0
            result['output'] = process.stdout
            result['error'] = process.stderr
            result['duration'] = time.time() - start_time
            status = "成功" if result['success'] else "失败"
            logging.info(f"{case_file.name}: {status} ({result['duration']:.2f}s)")
        except subprocess.TimeoutExpired:
            result['error'] = f"超时 ({timeout}s)"
            logging.warning(f"{case_file.name}: 超时")
        except Exception as e:
            result['error'] = str(e)
            logging.error(f"{case_file.name}: {str(e)}")
        return result
    def run_all_cases(self, parallel=False, max_workers=4):
        """运行所有案例"""
        case_files = list(self.cases_dir.glob("*.py"))
        if not case_files:
            print(f"在 {self.cases_dir} 中没有找到Python文件")
            return []
        print(f"找到 {len(case_files)} 个案例文件")
        print("=" * 50)
        results = []
        if parallel and len(case_files) > 1:
            # 并行运行
            with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
                future_to_file = {
                    executor.submit(self.run_single_case, f): f 
                    for f in case_files
                }
                for future in concurrent.futures.as_completed(future_to_file):
                    result = future.result()
                    results.append(result)
                    self.print_result(result)
        else:
            # 串行运行
            for case_file in case_files:
                print(f"\n运行: {case_file.name}")
                print("-" * 30)
                result = self.run_single_case(case_file)
                results.append(result)
                self.print_result(result)
        # 生成报告
        self.generate_report(results)
        return results
    def print_result(self, result):
        """打印运行结果"""
        status_icon = "✓" if result['success'] else "✗"
        print(f"{status_icon} {result['file'].name}")
        print(f"  耗时: {result['duration']:.2f}s")
        if result['error']:
            print(f"  错误: {result['error'][:200]}")
    def generate_report(self, results):
        """生成运行报告"""
        total = len(results)
        success = sum(1 for r in results if r['success'])
        failed = total - success
        total_time = sum(r['duration'] for r in results)
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        report = f"""
{'='*50}
批量运行报告
时间: {timestamp}
{'='*50}
总案例数: {total}
成功: {success}
失败: {failed}
总耗时: {total_time:.2f}s
平均耗时: {total_time/total:.2f}s if total > 0 else 0
{'='*50}
"""
        print(report)
        logging.info(report)
# 使用示例
if __name__ == "__main__":
    runner = BatchCaseRunner(cases_dir=".", parallel=True)
    results = runner.run_all_cases()
    # 也可以运行指定的案例
    # selected_cases = ["case1.py", "case2.py"]
    # for case in selected_cases:
    #     result = runner.run_single_case(Path(case))
    #     runner.print_result(result)

使用建议

选择合适的方法:
- 少量案例：使用简单的subprocess
- 大量案例：使用并行处理
- 测试场景：使用unittest/pytest
- 需要日志：使用完整的BatchCaseRunner
错误处理:
- 添加超时机制
- 捕获所有异常
- 记录详细的错误信息
性能优化:
- 使用并行处理加速
- 限制并发数量
- 监控内存使用