&nbsp;병렬로 API를 호출해서 같이 성능 테스트 또는&nbsp;Too Many Request 등을 TimedSemaphore 등으로 처리했을 때 테스트를 해볼 수 있다.&nbsp;&nbsp;<div style="background-color:#282c34;color:#abb2bf;"><pre style="font-family:'JetBrains Mono',monospace;font-size:9.4pt;">import aiohttp import asyncio import time from datetime import datetime import json async def send_request(session, method, url, headers, data, thread_id): start_time = time.time() start_datetime = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f') try: async with session.request(method=method, url=url, headers=headers, data=data) as response: elapsed_time = time.time() - start_time response_headers = dict(response.headers) try: response_body = await response.json() except aiohttp.ContentTypeError: response_body = await response.text() return { 'thread_id': thread_id, 'start_time': start_datetime, 'elapsed_time': elapsed_time, 'method': method, 'url': url, 'status_code': response.status, 'response_headers': response_headers, 'response_body': response_body } except Exception as e: elapsed_time = time.time() - start_time return { 'thread_id': thread_id, 'start_time': start_datetime, 'elapsed_time': elapsed_time, 'method': method, 'url': url, 'error': str(e) } async def main_async(): count = 100 # 요청 수 method = 'PUT' # 사용할 HTTP 메소드 url = 'http://localhost:8011/refresh' headers = { 'Content-Type': 'application/json', 'Authorization': 'Bearer jwtToken' } data = json.dumps({ "service": "DOMESTIC", "serviceType": "C2C", "sendAddress": { "zipCode": "06164", "baseAddress": "location", "detailAddress": "12층" }, "receiveAddress": { "zipCode": "06999", "baseAddress": "location", "detailAddress": "205동 402호" }, "productPrice": 10000, "limitWeight": 20000 }) async with aiohttp.ClientSession() as session: tasks = [ send_request(session, method, url, headers, data, i+1) for i in range(count) ] results = await asyncio.gather(*tasks) # 결과 출력 for result in sorted(results, key=lambda x: x['thread_id']): print(f"\n--- Thread {result['thread_id']} ---") print(f"Start Time: {result['start_time']}") print(f"Elapsed Time: {result['elapsed_time']:.4f} seconds") print(f"Method: {result['method']}") print(f"URL: {result['url']}") if 'error' in result: print(f"Error: {result['error']}") else: print(f"Status Code: {result['status_code']}") print("Response Headers:") for key, value in result['response_headers'].items(): print(f" {key}: {value}") print("Response Body:") if isinstance(result['response_body'], (dict, list)): print(json.dumps(result['response_body'], indent=2, ensure_ascii=False)) else: print(result['response_body']) # 결과 출력 이후: 스레드명 - 시작시간 (걸린시간) 요약 print("\n=== 요약 정보 ===") # 시작시간을 기준으로 오름차순 정렬 sorted_results = sorted(results, key=lambda x: x['start_time']) for result in sorted_results: thread_name = f"Thread {result['thread_id']:04}" start_time = result['start_time'] elapsed_time = f"{result['elapsed_time']:.4f} seconds" print(f"{thread_name} - {start_time} ({elapsed_time})") if __name__ == "__main__": asyncio.run(main_async())</pre></div>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;

[Python] 병렬 API 호출 예제

[Python] 병렬 API 호출 예제