test_api.py

"""https://zetcode.com/python/concurrent-http-requests/"""


import asyncio
import random
import time

import httpx

local_url = "http://127.0.0.1:5000"
remote_url = "https://cetinca-mathtext-nlu.hf.space/run/text2int_preprocessed"
remote_url = "https://tangibleai-mathtext.hf.space/run/"
headers = {"Content-Type": "application/json; charset=utf-8"}
data_list = [
    ["one hundred forty five"],
    ["ninety nine"],
    ["seventy three"],
    ["one thousand seven hundred sixty nine"],
]


# async call to endpoint
async def call_api(url, _data_list, number):
    data = random.choice(_data_list)
    json = {"data": data, "fn_index": 1}
    async with httpx.AsyncClient() as client:
        begin1 = time.time()
        begin2 = time.perf_counter()  # Used perf_counter for more precise result.
        # print(f"Call {number} started on: {start} text: {data}")
        response = await client.post(url=url, headers=headers, json=json, timeout=30)
        print(response.status_code)
        end1 = time.time()
        end2 = time.perf_counter()
        return f"Call_{number}: start: {begin1} end: {end1} delay: {end1 - begin1} with_time: {end2 - begin2}\n"\
               f"input_text: {data} result: {response.json().get('data')}"


async def main(n):
    calls = []
    for num in range(n):
        calls.append(call_api(remote_url, data_list, num))
    r = await asyncio.gather(*calls)
    print(*r, sep="\n")


asyncio.run(main(1))