如何使Python函数根据调用场景与参数自动切换并发/顺序执行？

现有如下Python函数：

def set_value(thing, set_quantity):
    set_function(thing, set_quantity) #sets "thing" to the desired quantity 
    read_value = read_function(thing) #reads quantity of "thing"
    if read_value != set_quantity:
        raise Exception('Values are not equal')

以及基于Robot Framework的测试用例（核心逻辑与测试框架无关）：

#test case 1
set_value('Bag_A', 5)

#Test case 2
set_value('Bag_B', 3)

#Test case 3
set_value('Bag_A', 2)
set_value('Bag_B', 8)
set_value('Purse_A',4)

期望实现的执行规则：

• 测试用例1、2中，单次调用set_value()需立即执行
• 测试用例3中，Bag_A、Bag_B属于同一类别（Bag类），二者的set_value()调用需并发执行；待该类别任务全部完成后，再顺序执行Purse_A（Purse类，与Bag不同类）的set_value()调用
• 物品类别无预定义限制，可任意扩展，且无法修改测试用例的编写方式

核心思路是给set_value做一层包装，通过任务缓存+类别分组+异步并发实现需求，完全兼容原有测试用例的调用逻辑：

1. 类别解析逻辑 #

从物品名称中提取类别前缀（假设命名规则为类别_编号，如Bag_A提取Bag），可通过字符串分割实现：

def get_category(thing):
    return thing.split('_')[0] if '_' in thing else thing

2. 任务缓存与执行控制 #

用全局变量维护当前类别、按类别分组的任务队列，在每次调用set_value时判断是否需要触发并发执行：

• 若当前任务与上一个任务类别相同，缓存任务参数
• 若类别不同，先并发执行上一类别所有缓存任务，再缓存当前任务
• 测试用例执行结束时，需手动触发剩余任务的执行（避免最后一组任务被遗漏）

3. 异步并发实现 #

将原set_value改造成异步函数，利用asyncio实现同类别任务的并发执行：

完整代码实现：

import asyncio

# 全局缓存：存储按类别分组的任务参数
_task_cache = {}
# 记录上一次调用的物品类别
_last_category = None

def get_category(thing):
    """从物品名称提取类别"""
    return thing.split('_')[0] if '_' in thing else thing

async def async_set_value(thing, set_quantity):
    """异步版本的set_value，保留原有业务逻辑"""
    set_function(thing, set_quantity)
    read_value = read_function(thing)
    if read_value != set_quantity:
        raise Exception('Values are not equal')

def set_value(thing, set_quantity):
    """包装后的set_value，实现任务缓存与并发控制"""
    global _last_category
    current_category = get_category(thing)
    
    # 1. 如果当前类别与上一个不同，先执行上一类别所有任务
    if _last_category is not None and current_category != _last_category:
        # 并发执行上一类别所有任务
        asyncio.run(asyncio.gather(*[
            async_set_value(*args) for args in _task_cache[_last_category]
        ]))
        # 清空已执行的类别缓存
        del _task_cache[_last_category]
    
    # 2. 将当前任务加入对应类别的缓存
    if current_category not in _task_cache:
        _task_cache[current_category] = []
    _task_cache[current_category].append((thing, set_quantity))
    
    # 3. 更新最后类别记录
    _last_category = current_category

def flush_remaining_tasks():
    """测试用例结束时调用，执行剩余未触发的任务"""
    global _last_category
    if _last_category is not None and _last_category in _task_cache:
        asyncio.run(asyncio.gather(*[
            async_set_value(*args) for args in _task_cache[_last_category]
        ]))
        _task_cache.clear()
        _last_category = None

4. 适配Robot Framework #

在Robot测试用例的Test Teardown阶段调用flush_remaining_tasks()，确保每个测试用例结束后都能执行剩余的缓存任务：

*** Settings ***
Test Teardown    flush_remaining_tasks

*** Test Cases ***
Test Case 1
    set_value    Bag_A    5

Test Case 2
    set_value    Bag_B    3

Test Case 3
    set_value    Bag_A    2
    set_value    Bag_B    8
    set_value    Purse_A    4

关键说明 #

• 无需修改原有测试用例的调用方式，完全兼容现有代码
• 自动识别任意新增类别，无需预定义类别列表
• 同类别任务会被缓存并在遇到不同类别时触发并发，符合测试用例3的执行要求
• 测试用例结束时的flush_remaining_tasks()确保最后一组任务不会被遗漏

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