test_api.py

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

import asyncio
import random
import time

import httpx

headers = {"Content-Type": "application/json; charset=utf-8"}

data_list_local = [
    {"url": "http://127.0.0.1:7860/run/text2int", "data": ["one hundred forty five", "text2int"]},
    {"url": "http://127.0.0.1:7860/run/text2int", "data": ["twenty thousand nine hundred fifty", "text2int_preprocessed"]},
    {"url": "http://127.0.0.1:7860/run/text2int", "data": ["one hundred forty five", "text2int"]},
    {"url": "http://127.0.0.1:7860/run/text2int", "data": ["nine hundred eighty three", "text2int_preprocessed"]},
    {"url": "http://127.0.0.1:7860/run/text2int", "data": ["five million"]},
    {"url": "http://127.0.0.1:7860/run/sentiment-analysis", "data": ["Totally agree"]},
    {"url": "http://127.0.0.1:7860/run/sentiment-analysis", "data": ["I like it"]},
    {"url": "http://127.0.0.1:7860/run/sentiment-analysis", "data": ["No more"]},
    {"url": "http://127.0.0.1:7860/run/sentiment-analysis", "data": ["I am not sure"]},
    {"url": "http://127.0.0.1:7860/run/sentiment-analysis", "data": ["Never"]},
]

data_remote = [
    {"url": "https://tangibleai-mathtext.hf.space/run/text2int", "data": ["one hundred forty five", "text2int"]},
    {"url": "https://tangibleai-mathtext.hf.space/run/text2int", "data": ["twenty thousand nine hundred fifty", "text2int_preprocessed"]},
    {"url": "https://tangibleai-mathtext.hf.space/run/text2int", "data": ["one hundred forty five", "text2int"]},
    {"url": "https://tangibleai-mathtext.hf.space/run/text2int", "data": ["nine hundred eighty three", "text2int_preprocessed"]},
    {"url": "https://tangibleai-mathtext.hf.space/run/text2int", "data": ["five million"]},
    {"url": "https://tangibleai-mathtext.hf.space/run/sentiment-analysis", "data": ["Totally agree"]},
    {"url": "https://tangibleai-mathtext.hf.space/run/sentiment-analysis", "data": ["I like it"]},
    {"url": "https://tangibleai-mathtext.hf.space/run/sentiment-analysis", "data": ["No more"]},
    {"url": "https://tangibleai-mathtext.hf.space/run/sentiment-analysis", "data": ["I am not sure"]},
    {"url": "https://tangibleai-mathtext.hf.space/run/sentiment-analysis", "data": ["Never"]},
]


# async call to endpoint
async def call_api(url, data, number):
    json = {"data": data}
    async with httpx.AsyncClient() as client:
        start = time.perf_counter()  # Used perf_counter for more precise result.
        response = await client.post(url=url, headers=headers, json=json, timeout=30)
        end = time.perf_counter()
        # print(response.status_code)
        return f"Call_{number}\n" \
               f"start: {start:.4f} end: {end:.4f} delay: {(end - start):.4f}\n" \
               f"input_text: {data}\n" \
               f"result: {response.json().get('data')}"


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


asyncio.run(main(30))