api_tester.py

# api_tester.py

import time
import asyncio
import urllib.request
import urllib.error

class APIBenchMarkTool:
    def __init__(self, target_url, total_request, concurrency):
        self.target_url = target_url
        self.total_request = total_request
        self.concurrency = concurrency
        self.result = []

    def _send_request(self):
        start_time = time.perf_counter()

        try:
            req = urllib.request.Request(
                self.target_url,
                headers={'User-Agent': 'Python-Benchmark-Tool/1.0'}
            )

            with urllib.request.urlopen(req, timeout=10) as response:
                latency = time.perf_counter() - start_time
                return {'success': True, 'status': response.status, 'latency': latency}

        except urllib.error.HTTPError as e:
            latency = time.perf_counter() - start_time
            return {'success': False, 'status': e.code, 'latency': latency}

        except Exception:
            latency = time.perf_counter() - start_time
            return {'success': False, 'status': 0, 'latency': latency}

    async def _worker(self, queue):
        loop = asyncio.get_event_loop()
        while not queue.empty():
            _ = await queue.get()
            result = await loop.run_in_executor(None, self._send_request)
            self.result.append(result)
            queue.task_done()

    async def run(self):
        print(f"\n[START] Benchmarking Target: {self.target_url}")
        print(f"Total Requests: {self.total_request} | Concurrency Level: {self.concurrency}\n")

        queue = asyncio.Queue()

        for i in range(self.total_request):
            queue.put_nowait(i)

        start_total_time = time.perf_counter()
        tasks = [asyncio.create_task(self._worker(queue)) for _ in range(self.concurrency)]

        await queue.join()

        for task in tasks:
            task.cancel()

        total_duration = time.perf_counter() - start_total_time