The Timer Class in Python’s threading Module

Greetings, everyone! In this article, we will explore the concepts of threads and timers, along with their specific features. Additionally, we will learn how to create our own custom timers that will execute code exactly the way we need.

Thread Theory

The Essence of the threading Module

The threading module provides the ability to create so-called threads in the code and manage them. A thread allows the program to efficiently distribute responsibilities for executing specific tasks, meaning it provides the ability to perform multiple tasks in less time.

What are Timers in threading?

Timers are a type of thread that we create to execute multiple processes simultaneously by setting a required execution time for a specific task.

This can be easily verified in code:

import threading

timer_is_thread = issubclass(threading.Timer, threading.Thread)
print(timer_is_thread)

Result:

True

Here, we compare the Timer class with the Thread class, which characterizes threads. According to the result, Timer is a subclass of Thread, and therefore, it is also a thread.

Example of a Standard Program Without Threads vs. With Them

Let’s look at an example. Suppose we have a program that receives input data from a user (a username and password), checks if the user entered the data correctly, and then checks whether such data already exists in a database.

Standard code works sequentially: first, it lets the user enter data, then it verifies the correctness of all the entered data, and only after that does it check if all this data is in the database. The entire process of entering and analyzing data is sequential.

So, our program can essentially be broken down into separate small processes:

1. Data entry.
2. Verification of data entry correctness.
3. Database information check.

The essence of threads implies that we create three threads for each process in the program. This way, our program can easily combine the execution of multiple processes simultaneously (although technically the code does not execute simultaneously, it just seems that way to us).

Code using threads is capable of executing all these processes without waiting for one of them to fully complete its work and let another start. This code simply uses the created threads to execute multiple processes at once. Thus, the database check process will not wait for the data entry process to finish verifying all of the user’s input.

In this case, after verifying the correctness of the entered username, the program will immediately start checking if this name is not in the database, while the user still hasn’t entered their password.

Advantages of Using Threads in Code

The main advantages of using threads are the speed and efficiency of code execution, as threads allow you to spend significantly less time performing various tasks by executing multiple processes simultaneously.

Creating Timers

A Timer That Starts Executing a Function After a Specified Time Interval

First, we need to import the threading module (it comes built-in, so there is no need to install it):

import threading

Let’s create the login() function:

def login():
    username = input("Username: ")
    password = input("Password: ")

    if username == "William" and password == "qwerty123":
        print("Welcome!")

We create a timer where we specify that the login() function will start executing only after 5 seconds (first, we pass the time interval, and then the function itself):

timer = threading.Timer(5.0, login)

After that, we absolutely must start the timer:

timer.start()

Here is the full code:

import threading

def login():
    first_name = input("Name: ")
    username = input("Username: ")
    password = input("Password: ")

    if first_name == "William" and username == "@w1lly22" and password == "qwerty123":
        print("Welcome!")

timer = threading.Timer(5.0, login)
timer.start()

Now, if you run this code, the login() function will indeed launch only after 5 seconds.

The First Method to Create a Repeated Timer

But is it possible to make the timer force a function to execute repeatedly at every time interval? Yes, but when exploring this issue, we encounter a specific feature of threads in Python.

This feature is that a thread can only be started once, so if we use a loop, we will get an error. For example, let’s try this code:

import threading

def no_user_found(first_names: list, usernames: list):
    for first_name, username in zip(first_names, usernames):
        print(f"First name: {first_name}, username: {username}")

        if first_name == "William" and username == "@w1lly22":
            print("The user with these data exits.")
        else:
            print("No user found.")

timer = threading.Timer(5.0, no_user_found)

while True:
    timer.start()

Result:

RuntimeError: threads can only be started once

To make a repeating timer, we should heed the essence of the error and start the thread only once. However, we will need to create our own separate class that will inherit from the Thread object.

To solve this problem, we first need to create our own class:

class RepeatedTimer(threading.Thread):
    def __init__(self, event):
        threading.Thread.__init__(self)
        self.stopped = event

It is important to know two things here:

1. The RepeatedTimer class inherits from the Thread class.
2. The self.stopped field is equal to the event variable, which takes a value of True or False and characterizes the event of executing and stopping the RepeatedTimer.

Next, we need to create a method that repeats the launches of our function:

def run(self):
    while not self.stopped.wait(5.0):
        no_user_found(first_names=["Alice", "Peter", "Brian", "William"],
                      usernames=["@aliceee", "@petermorton", "@brianllars", "@w1lly22"])

Here lies the main logic of the timer. As we know, a while loop executes as long as the condition is true. So, when the program starts, the event attribute and, accordingly, the self.stopped field will be False, and not False equals True.

Get Magnifique numérique’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

The next step is to start the created timer:

stop_flag = threading.Event()
custom_thread = RepeatedTimer(stop_flag)
custom_thread.start()

Here we declare the stop_flag object, which will be exactly the event attribute accepted by our RepeatedTimer class.

After 10 seconds, we change the value of the stop_flag object to True, and since not True equals False, the while loop will stop immediately.

stop_flag = threading.Event()
custom_thread = RepeatedTimer(stop_flag)
custom_thread.start()

time.sleep(10)
stop_flag.set()

This code will run for exactly 10 seconds, and then our stop_flag object will stop the program.

Full Code of the First Method

Here is the full code for the first method of creating a repeating timer:

import threading
import time

class RepeatedTimer(threading.Thread):
    def __init__(self, event):
        threading.Thread.__init__(self)
        self.stopped = event

    def run(self):
        while not self.stopped.wait(5.0):
            no_user_found(first_names=["Alice", "Peter", "Brian", "William"],
                          usernames=["@aliceee", "@petermorton", "@brianllars", "@w1lly22"])

stop_flag = threading.Event()
custom_thread = RepeatedTimer(stop_flag)
custom_thread.start()

time.sleep(10)
stop_flag.set()

The Second Method to Create a Repeated Timer

Nevertheless, there is another way to create a repeating timer. It differs in that instead of the Thread class, our class will specifically inherit from the Timer class.

Here is our function created earlier:

def no_user_found(first_names: list, usernames: list):
    for first_name, username in zip(first_names, usernames):
        print(f"First name: {first_name}, username: {username}")

        if first_name == "William" and username == "@w1lly22":
            print("The user with these data exits.")
        else:
            print("No user found.")

Here is the code for the second variant of creating a repeating timer:

class RepeatedTimer(threading.Timer):
    def run(self):
        while not self.finished.wait(self.interval):
            self.function(*self.args, **self.kwargs)

The logic here is exactly the same as in the previous variant, because self.finished, which is in the condition, is an object of the Event() class that we used in the last example. The self.interval variable is the time interval after which our self.function will be repeated.

Then everything becomes clearer, because initially, the condition in the while loop is not False, i.e., True. Therefore, as long as the condition is true, the function will repeat over and over again.

The body of the loop in this case is the line self.function(*self.args, **self.kwargs), which represents the function we will pass into the RepeatedTimer class. Meanwhile, *args and **kwargs are special parameters that help adapt this code to absolutely any functions we might pass into RepeatedTimer, as no one can predict what function and with how many parameters a hypothetical developer will want to pass.

It is thanks to the asterisks (* and **) that our repeating timer will work regardless of the number of parameters we set for the no_user_found() function. (If you haven't encountered the *args and **kwargs parameters before, I recommend reading about them in the section below).

Next, we create a timer object using our RepeatedTimer class, where we first pass the time interval for repeating the function and the no_user_found function itself.

It is worth noting that the second implementation of the repeating timer excludes writing parameters into the function the way we did before, namely:

no_user_found(first_names=["Alice", "Peter", "Brian", "William"],
              usernames=["@aliceee", "@petermorton", "@brianllars", "@w1lly22"])

Instead, we need to specify the parameter values in the form of a dictionary and pass it to the kwargs parameter, which handles dictionaries:

timer = RepeatedTimer(1, no_user_found, kwargs={
    "first_names": ["Alice", "Peter", "Brian", "William"],
    "usernames": ["@aliceee", "@petermorton", "@brianllars", "@w1lly22"]
    })

timer.start()

Now let’s briefly break down what we passed into the RepeatedTimer() class:

1. First, we pass the interval parameter.
2. Next, we pass the no_user_found function itself.
3. A dictionary containing lists.

We stop the timer (and accordingly, the function execution) after 10 seconds:

time.sleep(10)
timer.cancel()

Also, it should be understood that our function will repeat precisely during these 10 seconds, until we cancel the timer.

Full Code of the Second Method

Here is the full code for the second method of creating a repeating timer:

import threading
import time

class RepeatedTimer(threading.Timer):
    def run(self):
        while not self.finished.wait(self.interval):
            self.function(*self.args, **self.kwargs)

def no_user_found(first_names: list, usernames: list):
    for first_name, username in zip(first_names, usernames):
        print(f"First name: {first_name}, username: {username}")

        if first_name == "William" and username == "@w1lly22":
            print("The user with these data exits.")
        else:
            print("No user found.")

timer = RepeatedTimer(1, no_user_found, kwargs={
    "first_names": ["Alice", "Peter", "Brian", "William"],
    "usernames": ["@aliceee", "@petermorton", "@brianllars", "@w1lly22"]
    })

timer.start()

time.sleep(10)
timer.cancel()

Special Parameters *args and kwargs

In Python, the *args notation stands for positional arguments, meaning lists, sets, etc. It is important to understand here that only the asterisks are crucial, not the words args or kwargs themselves. The asterisks (* or **) are responsible for correctly unpacking the elements of the data types presented in this section.

When we pass such data types into a function as parameters, a single asterisk (*) extracts all the elements in these data types and provides them to the function as regular comma-separated values.

Here is an example:

user_info = ["John", "Peterson", "@johnnypi", "admin"]

def account_private_info(first_name, last_name, username, password):
    print(f"USER INFO\n"
          f"============================\n"
          f"First name:   {first_name}\n"
          f"Last name:    {last_name}\n"
          f"Username:     {username}\n"
          f"Password:     {password}")

account_private_info(*user_info)

Result:

USER INFO
============================
First name:   John
Last name:    Peterson
Username:     @johnnypi
Password:     admin

As we can see, using a single asterisk, we were able to easily break down all the list elements into the individual parameters specified during the function declaration.

This code yields the same result as this notation:

user_info = ["John", "Peterson", "@johnnypi", "admin"]

def account_private_info(first_name, last_name, username, password):
    print(f"USER INFO\n"
          f"============================\n"
          f"First name:   {first_name}\n"
          f"Last name:    {last_name}\n"
          f"Username:     {username}\n"
          f"Password:     {password}")

account_private_info(user_info[0], user_info[1], user_info[2], user_info[3])

It’s just that the first notation is much more convenient thanks to the clean unpacking of the list elements. Here, we had to manually write out the values of all parameters when calling the function.

The **kwargs notation stands for keyword arguments, meaning dictionaries. Otherwise, two asterisks (**) do the same thing as one, but they work with different data types.

Example:

user_info = {"first_name": "John",
             "last_name": "Peterson",
             "username": "@johnnypi",
             "password": "admin"}

def account_private_info(first_name, last_name, username, password):
    print(f"USER INFO\n"
          f"============================\n"
          f"First name:   {first_name}\n"
          f"Last name:    {last_name}\n"
          f"Username:     {username}\n"
          f"Password:     {password}")

account_private_info(**user_info)

This code is equivalent to the following notation:

user_info = {"first_name": "John",
             "last_name": "Peterson",
             "username": "@johnnypi",
             "password": "admin"}

def account_private_info(first_name, last_name, username, password):
    print(f"USER INFO\n"
          f"============================\n"
          f"First name:   {first_name}\n"
          f"Last name:    {last_name}\n"
          f"Username:     {username}\n"
          f"Password:     {password}")

account_private_info(user_info["first_name"], user_info["last_name"], user_info["username"], user_info["password"])

Again, the first variant using two asterisks is much better.

Conclusion

In this article, we got acquainted with the threading module and the Timer class. We also reviewed in detail two options for creating a repeating timer.

Sources

1. Code for the methods of creating a repeating timer.