To improve the given script using asyncio, you can use the aiohttp library for making asynchronous HTTP requests. Here’s the modified version of the script:
#!/usr/bin/env python3
import json
import asyncio
import aiohttp
import time
from typing import List, Dict
TIME_BETWEEN_REQUESTS = 5
TIME_TOTAL = 60
async def get_latency(session, domain):
try:
start = time.time()
if not domain.startswith(("http://", "https://")):
domain = "https://" + domain
async with session.get(domain, timeout=3) as response:
end = time.time()
return end - start
except asyncio.TimeoutError:
return float("inf")
except aiohttp.client_exceptions.ServerDisconnectedError:
return float("inf")
def add_latency_to_domain(domain, latency, latencies):
if domain not in latencies:
latencies[domain] = []
latencies[domain].append(latency)
return latencies
async def measure_latencies_for_domains(session, domains, latencies):
tasks = []
for domain in domains:
tasks.append(get_latency(session, domain))
results = await asyncio.gather(*tasks)
for domain, latency in zip(domains, results):
latencies = add_latency_to_domain(domain, latency, latencies)
return latencies
async def measure_latencies(domains, duration):
latencies = {}
start_time = time.time()
end_time = start_time + duration
async with aiohttp.ClientSession() as session:
while time.time() < end_time:
latencies = await measure_latencies_for_domains(session, domains, latencies)
await asyncio.sleep(TIME_BETWEEN_REQUESTS)
return latencies
def average_latencies(latencies):
averages = []
for domain, latency_list in latencies.items():
avg_latency = sum(latency_list) / len(latency_list)
averages.append((domain, avg_latency))
return averages
def sort_latencies(averages):
return sorted(averages, key=lambda x: x[1])
async def get_latency_report(domains, duration):
latencies = await measure_latencies(domains, duration)
averages = average_latencies(latencies)
return sort_latencies(averages)
def get_instances(data: Dict) -> List[Dict]:
instances = []
for instance_details in data["instance_details"]:
instances.append(instance_details)
return instances
def get_domains(instances: List[Dict]) -> List[str]:
return [instance["domain"] for instance in instances]
def load_json_data(filepath: str) -> Dict:
with open(filepath) as json_data:
return json.load(json_data)
async def main():
data = load_json_data('stats.json')
instances = get_instances(data)
domains = get_domains(instances)
report = await get_latency_report(domains, TIME_TOTAL)
for domain, avg_latency in report:
print(f"{domain}: {avg_latency:.2f} seconds")
if __name__ == "__main__":
asyncio.run(main())
This version of the script should be able to ping multiple instances simultaneously using asyncio.
From the name I thought it was about chatgpt prompts or something, but I always something much more interesting whenever I see a post. what exactly is the community about?
Whenever I ask anything to the AI if I think the response is helpful I post it here. This is just the AI output.
Ah in that case I think it would be better if you mention the prompt asked in the body post, so users have a better idea of the context.