-1

I'm planning to call multiple functions and execute SQL queries from random inputs from JSON.

Actual code:

def daily():
    db_connection = DatabaseConnection()

    aaa(db_connection)
    bbb(db_connection)

    return {'success': True}


def aaa(db_connection):
    database_table = 'aaa'
    symbols_list = [{'code': 'XXX', 'database_column': 'aaa_1'}, {'code': 'YYY', 'database_column': 'aaa_2'}]
    load_data_from_api_to_database(db_connection, database_table, symbols_list)


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})

        json_data = json.loads(response.text)

        if response.status_code != 200:
            return

        data_points = json_data['dataPoint']
        for x, _ in enumerate(data_points):
            value = data_points[x]['value']
            date = data_points[x]['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()

bbb() similar to aaa() just different json array value.

Test code:

class TestDailyHandler(unittest.TestCase):
    @classmethod
    def setup_class(cls):
        cls.mock_get_patcher = patch('src.daily_handler.HttpRequest.get')
        cls.mock_get = cls.mock_get_patcher.start()
        cls.mock_database_connection_patcher = patch('src.daily_handler.DatabaseConnection')
        cls.mock_database_connection = cls.mock_database_connection_patcher.start()

    def load_data_from_api_to_database(self):
        assert load_data_from_api_to_database({}, None) == {'success': True}

        symbols_list = [{'code': 'XXX', 'database_column': 'aaa_1'}, {'code': 'YYY', 'database_column': 'aaa_2'}]

        for x in range(len(symbols_list)):
            code = [x]['code']
            self.mock_get.assert_any_call(f'https://api.example.com/value/symbols_list{code}', headers={'accept': 'application/json', 'appkey': self.mock_get_aws_secret.return_value})

        db_execute_many_args_list = self.mock_database_connection.return_value.execute_many.call_args_list
        daily_table_insert_command_length = len([x for x in db_execute_many_args_list if re.search(r'INSERT INTO ', str(x), re.IGNORECASE)])

        self.assertEqual(daily_table_insert_command_length, len(db_execute_many_args_list))
        self.assertEqual(self.mock_database_connection.return_value.commit.call_count, daily_table_insert_command_length)

        db_execute_many_args_list = self.mock_database_connection.return_value.execute_many.call_args_list
        daily_table_update_command_length = len([x for x in db_execute_many_args_list if re.search(r'UPDATE ', str(x), re.IGNORECASE)])

        self.assertEqual(daily_table_update_command_length, len(db_execute_many_args_list))
        self.assertEqual(self.mock_database_connection.return_value.commit.call_count, daily_table_update_command_length)

By the way, I'm not sure how to call multiple functions aaa() and bbb(). I supposely to test starting from daily() instead load_data_from_api_to_database() function. Also JSON array input from each function. Currently it's static value.

halfer
  • 19,824
  • 17
  • 99
  • 186
Nurdin
  • 23,382
  • 43
  • 130
  • 308
  • Hi, I think you can remove those `enumerate` as it is pretty unnecessary. Use `requests` module and it comes with nifty `raise_for_status` method to throw error when you get 4xx or 5xx. Just use parametrize to set your `symbols_list` test data. Happy unit testing. – Sian Lerk Lau Nov 30 '21 at 13:51

1 Answers1

0

Warning: the following code is provided as guidance only

taken from example I have a string whose content is a function name, how to refer to the corresponding function in Python?

additional reading Store functions in list and call them later

new answer


def aaa(db_connection, data):
    # bind from data
    symbols_list = data
    # ... 

def bbb(db_connection, data):
    # bind from data
    symbols_list = data
    # ...

dispatcher = { "aaa" : aaa, "bbb" : bbb }

def daily():
    db_connection = DatabaseConnection()

    #loop the dict for functions
    for fn in dispatcher:
        if(callable(dispatcher[func_name]))
            json_str = input('Enter your JSON data:')
            try
                data = json.load(json_str);
                dispatcher[func_name](db_connection, data)
            except JSONDecodeError
                print('Error loading json')
                return None

    return {'success': True}
Brian Wong
  • 11
  • 2