Skip to content

Instantly share code, notes, and snippets.

@seanpianka
Created April 28, 2018 18:04
Show Gist options
  • Save seanpianka/8d911bc1403b61f8d1b66c9a27812634 to your computer and use it in GitHub Desktop.
Save seanpianka/8d911bc1403b61f8d1b66c9a27812634 to your computer and use it in GitHub Desktop.
Find all usernames that are (potentially available) to use on GitHub
from queue import Queue
from threading import Thread, Lock
from itertools import product
from string import ascii_lowercase
import requests
usrs = [''.join(i) for i in product(ascii_lowercase + '-', repeat = 3)]
usrs = [x for x in usrs if (not x.startswith('-')) and (not x.endswith('-'))]
class Worker(Thread):
def __init__(self, tasks):
"""
:param tasks: A queue containing the tasks for the worker instance.
:type tasks: queue.Queue instance
"""
Thread.__init__(self)
self.tasks = tasks
self.daemon = True
self.start()
def run(self):
while True:
func, args, kwargs = self.tasks.get()
while True:
try:
func(*args, **kwargs)
except Exception as e:
print(e)
continue
else:
self.tasks.task_done()
break
class ThreadPool:
def __init__(self, thread_count):
self.tasks = Queue(thread_count)
# underscore because unused counter variable
for _ in range(thread_count):
Worker(self.tasks)
def add_task(self, func, *args, **kwargs):
""" Add a new task to the queue. """
self.tasks.put((func, args, kwargs))
def map(self, func, args_list):
""" Add a list of extant tasks to queue. """
for args in args_list:
self.add_task(func, args)
def wait_completion(self):
""" Blocks until all tasks in queue have been processed. """
self.tasks.join()
# thread pool with n Workers
pool = ThreadPool(15)
lock = Lock()
def check_available(username):
if requests.get('https://github.com/{}'.format(username)).status_code == 404:
print(username + ' is available.')
for username in usrs:
pool.add_task(check_available, username)
pool.wait_completion()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment