如何基于JSON随机输入调用多个函数并执行SQL查询

一、业务代码调整（支持动态JSON输入、多函数统一调度） #

原有aaa()、bbb()逻辑重复，我们直接抽离公共配置，既不需要重复写相似函数，也支持外部传入随机JSON数组作为输入：

import json
import random
# 替换为你实际依赖的导入路径
from your_module import DatabaseConnection, HttpRequest, api_key, database_date_column

# 原有aaa、bbb的静态配置抽离，新增同类逻辑仅需加配置项即可
TASK_CONFIG = {
    "aaa": [{'code': 'XXX', 'database_column': 'aaa_1'}, {'code': 'YYY', 'database_column': 'aaa_2'}],
    "bbb": [{'code': 'ZZZ', 'database_column': 'bbb_1'}, {'code': 'UUU', 'database_column': 'bbb_2'}]
}

def daily(custom_config: dict = None):
    """
    主入口函数，支持传入自定义随机配置，不传则使用默认静态配置
    :param custom_config: 格式与TASK_CONFIG一致
    """
    db_connection = DatabaseConnection()
    task_config = custom_config if custom_config else TASK_CONFIG
    
    # 统一调度所有配置对应的任务，不需要单独调用aaa、bbb
    for table_name, symbols_list in task_config.items():
        load_data_from_api_to_database(db_connection, table_name, symbols_list)

    return {'success': True}


def load_data_from_api_to_database(db_connection, database_table, symbols_list):
    http_request = HttpRequest()
    for _, symbol in enumerate(symbols_list):
        code = symbol['code']
        database_column = symbol['database_column']

        response = http_request.get(
            f'https://api.example.com/value/{code}', 
            headers={'accept': 'application/json', 'appkey': api_key}
        )

        # 单个请求失败跳过，不终止后续其他任务，比直接return更合理
        if response.status_code != 200:
            continue
        try:
            json_data = json.loads(response.text)
        except json.JSONDecodeError:
            continue

        data_points = json_data.get('dataPoint', [])
        for point in data_points:
            value = point['value']
            date = point['date']
            db_connection.execute(
                f'INSERT INTO "{database_table}" ("{database_date_column}") VALUES (%(date_time)s) ON CONFLICT ("{database_date_column}") DO NOTHING', 
                {'date_time': date}
            )
            db_connection.execute(
                f'UPDATE "{database_table}" SET "{database_column}" = %(value)s WHERE "{database_date_column}" = %(date_time)s', 
                {'value': value, 'date_time': date}
            )
        db_connection.commit()

随机JSON输入生成示例 #

需要随机输入时直接生成自定义配置传入daily即可：

def generate_random_config(task_count=2, symbol_per_task=2):
    random_config = {}
    for i in range(task_count):
        table_name = f"test_table_{random.randint(1,100)}"
        symbols = []
        for j in range(symbol_per_task):
            symbols.append({
                "code": ''.join(random.choices('ABCDEFGHIJKLMNOPQRSTUVWXYZ', k=3)),
                "database_column": f"test_col_{random.randint(1, 10)}"
            })
        random_config[table_name] = symbols
    return random_config

# 调用示例：传入随机配置执行业务逻辑
# daily(custom_config=generate_random_config())

二、测试代码修正（从daily入口测试） #

原有测试存在函数名不规范、语法错误、方法调用匹配错误的问题，修正后如下：

import unittest
from unittest.mock import patch
import re
import json
# 替换为你实际的函数导入路径
from src.daily_handler import daily, TASK_CONFIG, generate_random_config, api_key

class TestDailyHandler(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        # 模拟HTTP请求
        cls.mock_get_patcher = patch('src.daily_handler.HttpRequest.get')
        cls.mock_get = cls.mock_get_patcher.start()
        cls.mock_get.return_value.status_code = 200
        cls.mock_get.return_value.text = json.dumps({
            "dataPoint": [
                {"value": random.randint(1,100), "date": "2024-01-01"},
                {"value": random.randint(1,100), "date": "2024-01-02"}
            ]
        })
        # 模拟数据库连接
        cls.mock_db_patcher = patch('src.daily_handler.DatabaseConnection')
        cls.mock_db = cls.mock_db_patcher.start()
        cls.mock_db_instance = cls.mock_db.return_value

    @classmethod
    def tearDownClass(cls):
        cls.mock_get_patcher.stop()
        cls.mock_db_patcher.stop()

    def test_daily_with_default_config(self):
        # 直接从主入口调用测试
        result = daily()
        self.assertEqual(result, {'success': True})

        # 校验接口调用符合预期
        all_symbols = []
        for symbols in TASK_CONFIG.values():
            all_symbols.extend(symbols)
        expected_calls = [
            unittest.mock.call(
                f'https://api.example.com/value/{s["code"]}', 
                headers={'accept': 'application/json', 'appkey': api_key}
            ) for s in all_symbols
        ]
        self.mock_get.assert_has_calls(expected_calls, any_order=True)

        # 校验SQL执行符合预期
        execute_calls = self.mock_db_instance.execute.call_args_list
        insert_count = len([c for c in execute_calls if re.search(r'INSERT INTO ', str(c), re.IGNORECASE)])
        update_count = len([c for c in execute_calls if re.search(r'UPDATE ', str(c), re.IGNORECASE)])
        self.assertEqual(insert_count, update_count)
        # 每个symbol处理完提交一次，commit次数等于symbol总数
        self.assertEqual(self.mock_db_instance.commit.call_count, len(all_symbols))

    def test_daily_with_random_config(self):
        # 测试随机输入场景
        random_config = generate_random_config(task_count=3, symbol_per_task=2)
        result = daily(custom_config=random_config)
        self.assertEqual(result, {'success': True})

        all_symbols = []
        for symbols in random_config.values():
            all_symbols.extend(symbols)
        self.assertEqual(self.mock_get.call_count, len(all_symbols))

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