CODEWITHSUNDEEP
python
papelr
papelr

Reputation: 438

Python: Pinging a URL multiple times at once for testing

I have a link that I want to test for robustness, for lack of a better word. What I have code that pings the URL multiple times, sequentially:

# Testing for robustness
for i in range(100000):
    city = 'New York'
    city = '%20'.join(city.split(' '))

    res = requests.get(f'http://example.com/twofishes?query={city}')

    data = res.json()
    geo = data['interpretations'][0]['feature']['geometry']['center']
    print('pinging xtime: %s ' % str(i))
    print(geo['lat'], geo['lng'])

I want to take this code, but ping the link say, 10 or 12 times at once. I don't mind the sequential pinging, but that's not as efficient as pinging multiple times at once. I feel like this is a quick modification, where the for loop comes out and a PULL function goes in?

Upvotes: 0

Views: 581

Answers (1)

The Pineapple
The Pineapple

Reputation: 567

Here is an example program which should work for this task. Given that I do not want to be blacklisted, I have not actually tested the code to see if it works. Regardless, it should at least be in the ballpark of what your looking for. If you want actually have all of the threads execute at the same time I would look into adding events. Hope this helps.

Code

import threading
import requests
import requests.exceptions as exceptions


def stress_test(s):
    for i in range(100000):
        try:
            city = 'New York'
            city = '%20'.join(city.split(' '))

            res = s.get(f'http://example.com/twofishes?query={city}')

            data = res.json()
            geo = data['interpretations'][0]['feature']['geometry']['center']
            print('pinging xtime: %s ' % str(i))
            print(geo['lat'], geo['lng'])
        except (exceptions.ConnectionError, exceptions.HTTPError, exceptions.Timeout):
            pass


if __name__ == '__main__':
    for i in range(1, 12):
        s = requests.session()
        t = threading.Thread(target=stress_test, args=(s,))
        t.start()

    for th in threading.enumerate():
        if th != threading.current_thread():
            th.join()

Upvotes: 1

Related Questions