How to Extend Python Thread Class

Summary: in this tutorial, you will learn how to extend the Python Thread class to run code in a new thread.

Introduction to Python Thread Class

When a Python program starts, it has a thread called the main thread. Sometimes, you want to offload the I/O-bound tasks to a new thread to execute them concurrently. To do that, you use the built-in threading module.

One way to execute code in a new thread is to extend the Thread class of the threading module. Here are the steps:

• First, define a subclass of threading.Thread class.
• Second, override __init__(self, [,args]) method inside of the __init__() method of the subclass to add custom arguments.
• Third, override the run(self, [,args]) method inside of the subclass to customize the behavior of the new thread class when a new thread is created.

Let’s take an example of extending the Thread class. We’ll develop a class that performs an HTTP request to a URL and display the response code:

class HttpRequestThread(Thread):
    def __init__(self, url: str) -> None:
        super().__init__()
        self.url = url

    def run(self) -> None:
        print(f'Checking {self.url} ...')
        try:
            response = urllib.request.urlopen(self.url)
            print(response.code)
        except urllib.error.HTTPError as e:
            print(e.code)
        except urllib.error.URLError as e:
            print(e.reason)Code language: Python (python)

How it works.

First, define a HttpRequestThread that extends the Thread class from the threading module:

class HttpRequestThread(Thread):Code language: Python (python)

Second, define the __init__() method that accepts a URL. Inside the __init__() method calls the __init__() method of the superclass.

def __init__(self, url: str) -> None:
    super().__init__()
    self.url = urlCode language: Python (python)

Third, override the run method that uses the urllib to get the HTTP status code of the specified URL and display it to the console:

def run(self) -> None:
    print(f'Checking {self.url} ...')
    try:
        response = urllib.request.urlopen(self.url)
        print(response.code)
    except urllib.error.HTTPError as e:
        print(e.code)
    except urllib.error.URLError as e:
        print(e.reason)Code language: Python (python)

To use the HttpRequestThread class, you create instances of the HttpRequestThread class and call the start() method. Also, you can call the join() method to wait for all the threads to complete.

The following defines the main() function that uses the HttpRequestThread class:

def main() -> None:
    urls = [
        'https://httpstat.us/200',
        'https://httpstat.us/400'
    ]

    threads = [HttpRequestThread(url) for url in urls]

    [t.start() for t in threads]

    [t.join() for t in threads]Code language: Python (python)

How it works.

First, define a list of urls that we want to check:

urls = [
    'https://httpstat.us/200',
    'https://httpstat.us/400'
]Code language: Python (python)

Second, create instances of the HttpRequestThread based on the urls list using list comprehension. The list comprehension returns a list of instances of the HttpRequestThread class:

threads = [HttpRequestThread(url) for url in urls]Code language: Python (python)

Third, call the start() method of each thread in the threads list:

[t.start() for t in threads]Code language: Python (python)

Finally, call the join of each Thread instance to wait for all the threads to complete:

[t.join() for t in threads]Code language: Python (python)

Put it all together:

from threading import Thread
import urllib.request


class HttpRequestThread(Thread):
    def __init__(self, url: str) -> None:
        super().__init__()
        self.url = url

    def run(self) -> None:
        print(f'Checking {self.url} ...')
        try:
            response = urllib.request.urlopen(self.url)
            print(response.code)
        except urllib.error.HTTPError as e:
            print(e.code)
        except urllib.error.URLError as e:
            print(e.reason)


def main() -> None:
    urls = [
        'https://httpstat.us/200',
        'https://httpstat.us/400'
    ]

    threads = [HttpRequestThread(url) for url in urls]

    [t.start() for t in threads]

    [t.join() for t in threads]


if __name__ == '__main__':
    main()Code language: Python (python)

Output:

Checking https://httpstat.us/200 ...
Checking https://httpstat.us/400 ...
200
400Code language: JavaScript (javascript)

Summary

• Extend the Thread class to run code in a new thread by calling the __init__() method of the superclass in the subclass and override the run method to add the code to run in a new thread.