TangibleAI/mathtext-fastapi

1"""https://zetcode.com/python/concurrent-http-requests/"""2 3import asyncio4import random5import time6import pandas as pd7import httpx8from os.path import exists9 10NUMBER_OF_CALLS = 111 12headers = {"Content-Type": "application/json; charset=utf-8"}13 14# base_url = "https://tangibleai-mathtext-fastapi.hf.space/{endpoint}"15base_url = "http://localhost:7860/run/{endpoint}"16 17data_list_1 = {18    "endpoint": "text2int",19    "test_data": [20        "one hundred forty five",21        "twenty thousand nine hundred fifty",22        "one hundred forty five",23        "nine hundred eighty three",24        "five million",25    ]26}27 28data_list_2 = {29    "endpoint": "text2int-preprocessed",30    "test_data": [31        "one hundred forty five",32        "twenty thousand nine hundred fifty",33        "one hundred forty five",34        "nine hundred eighty three",35        "five million",36    ]37}38data_list_3 = {39    "endpoint": "sentiment-analysis",40    "test_data": [41        "Totally agree",42        "I like it",43        "No more",44        "I am not sure",45        "Never",46    ]47}48 49 50# async call to endpoint51async def call_api(url, data, call_number, number_of_calls):52    json = {"data": [data]}53    async with httpx.AsyncClient() as client:54        start = time.perf_counter()  # Used perf_counter for more precise result.55        response = await client.post(url=url, headers=headers, json=json, timeout=30)56        end = time.perf_counter()57        return {58            "endpoint": url.split("/")[-1],59            "test data": data,60            "status code": response.status_code,61            "response": response.json().get("data"),62            "call number": call_number,63            "number of calls": number_of_calls,64            "start": start.__round__(4),65            "end": end.__round__(4),66            "delay": (end - start).__round__(4)67        }68 69 70data_lists = [data_list_1, data_list_2, data_list_3]71 72results = []73 74 75async def main(number_of_calls):76    for data_list in data_lists:77        calls = []78        for call_number in range(1, number_of_calls + 1):79            url = base_url.format(endpoint=data_list["endpoint"])80            data = random.choice(data_list["test_data"])81            calls.append(call_api(url, data, call_number, number_of_calls))82        r = await asyncio.gather(*calls)83        results.extend(r)84 85 86 87start = time.perf_counter()88asyncio.run(main(NUMBER_OF_CALLS))89end = time.perf_counter()90print(end-start)91df = pd.DataFrame(results)92 93if exists("call_history.csv"):94    df.to_csv(path_or_buf="call_history.csv", mode="a", header=False, index=False)95else:96    df.to_csv(path_or_buf="call_history.csv", mode="w", header=True, index=False)97