threaded_requests.py

#!/usr/bin/python3
"""
A demonstration of how to perform multi-threaded HTTP requests.

Third Party Dependencies
------------------------
requests@latest:
    Required to perform HTTP requests (unless you want to use urllib).
"""
import requests
from concurrent.futures import ThreadPoolExecutor


def check_website_status(*urls: str, thread_count=4) -> tuple[bool, ...]:
    """Check the return status of a list of websites.

    This function will take a set of arguments and perform a HTTP POST request
    using a thread pool.

    Arguments
    ---------
    urls: str
        A starred expression which takes a series of positional parameter
        string URLs.
    thread_count: int, optional
        Sets how many threads/workers should be used at any one time.
        Default: 4

    Returns
    -------
    tuple[bool, ...]
        An immutable tuple of boolean values indicating if the return code was
        below 400. This tuple should always be the same length as the amount
        of given urls.
    """

    def worker(url: str) -> bool:
        """Checks the status code of a website

        Performs a HTTP HEAD request and returns a boolean depending on the
        status code returned

        Parameters
        ----------
        url: str
            The url to check

        Returns
        -------
        bool
            Returns true if the status code is below 400
            (i.e: The request was successful)

        Raises
        ------
        MissingSchema:
            Raised when an invalid URL string was provided.
        """
        response = requests.head(url)
        return response.ok  # Returns True if the status code is less than 400.

    # This list will keep track of "futures". Futures represent function which
    # may have not returned yet.
    futures = []

    # Define the thread pool which will manage applying workers to jobs.
    executor = ThreadPoolExecutor(max_workers=thread_count)

    for url in urls:
        future = executor.submit(worker, url)  # Submit a job to the pool.
        futures.append(future)  # Store the future in our list to track.

    # Fetch all results from futures and then store them in an immutable tuple.
    return tuple(future.result() for future in futures)


if __name__ == "__main__":
    results = check_website_status(
        "https://www.google.com/", "https://www.amazon.com/", "invalid"
    )

    print(results)