Skip to content

Instantly share code, notes, and snippets.

@sokol8
Forked from rednafi/concurrent_hit.py
Created February 11, 2024 22:10
Show Gist options
  • Save sokol8/e0a9f87e2f840a84514dbb12b3fe0698 to your computer and use it in GitHub Desktop.
Save sokol8/e0a9f87e2f840a84514dbb12b3fe0698 to your computer and use it in GitHub Desktop.
1000 concurrent API request with ThreadPoolExecutor Python
import time
from concurrent.futures import ThreadPoolExecutor
from functools import wraps
import requests
from tqdm import tqdm
def timeit(method):
@wraps(method)
def wrapper(*args, **kwargs):
start_time = time.time()
result = method(*args, **kwargs)
end_time = time.time()
print(f"{method.__name__} => {(end_time-start_time)*1000} ms")
return result
return wrapper
def attack_one(url):
payload = '{\n\t"query": "West Shaorapara,around Mirpur 10,\\nShapla sharani.\\nHouse no:438/3"\n}'
headers = {
"x-api-key": "1234",
"Content-Type": "application/json",
}
response = requests.request("GET", url, headers=headers, data=payload)
return response.text.encode("utf8")
@timeit
def attack_all(urls):
with ThreadPoolExecutor(max_workers=13) as executor:
results = list(
tqdm(executor.map(attack_one, urls, timeout=60), total=len(urls))
)
return results
if __name__ == "__main__":
url = "http://helloworld.tech/"
urls = [url] * 1000
results = attack_all(urls)
for result in results:
print(result)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment