Всіх вітаю! У цій статті ми розглянемо, як асинхронно запускати код через деякий проміжок часу за допомогою класу Timer вбудованого модуля threading.
Теорія потоків
Суть модуля threading
Модуль threading надає можливість створювати так звані потоки у коді та керувати ними. Потік дозволяє програмі працювати асинхронно, тобто це можливість виконувати багато задач за меншу кількість часу.
Що таке таймери у threading?
Таймери — це різновид потоків, які ми створюємо, щоб виконувати декілька процесів одночасно, виставляючи у них потрібний час виконання того чи іншого завдання.
Це можна легко перевірити у коді:
import threading
timer_is_thread = issubclass(threading.Timer, threading.Thread)
print(timer_is_thread)Результат:
TrueТут ми порівнюємо клас Timer з класом Thread, який характеризує потоки. Згідно з результатом, Timer є підкласом Thread, а отже, є й потоком.
Приклад звичайної програми без потоків і з ними
Наведемо приклад. Припустимо, у нас є програма, яка отримує дані входу від користувача (ім’я користувача та пароль), перевіряє те, наскільки правильно він ввів дані, а потім перевіряє, чи є такі дані вже у базі даних.
Стандартний код працює поступово: спочатку дає користувачу ввести дані, потім перевіряє правильність введення усіх даних і тільки після цього перевіряє, чи є усі ці дані у базі даних. Весь процес введення та аналізу даних є послідовним.
Тож нашу програму по суті можна розбити на окремі маленькі процеси:
Введення даних.
Перевірка правильності введення даних.
Перевірка інформації по базі даних.
Суть потоків якраз передбачає те, що ми створимо три потоки під кожний процес у програмі. Таким чином, наша програма зможе з легкістю поєднувати виконання декількох процесів одночасно (хоча технічно код все одно не виконується одночасно, просто нам так здається).
Код з використанням потоків здатний виконати усі ці процеси, не чекаючи, поки один із них повністю завершить свою роботу і дасть іншому почати. Цей код просто використає створені потоки для виконання одразу декількох процесів одночасно. Тоді процес перевірки по базі даних не буде чекати, поки процес перевірки правильності введення повністю перевірить усі дані користувача.
У такому разі після перевірки правильності введення імені користувача програма одразу почне перевіряти, чи немає цього імені у базі даних, допоки користувач все ще не ввів пароль.
Переваги використання потоків у коді
Головною перевагою використання потоків є швидкість виконання коду, адже потоки дозволяють витрачати значно менше часу на виконання тих чи інших задач завдяки одночасному виконанню декількох задач.
Створення таймерів
Таймер, що починає виконання функції через визначений проміжок часу
Отже, спочатку нам потрібно імпортувати модуль threading (він вже є у вас, а тому встановлювати його потреби немає):
import threadingСтворюємо функцію login():
def login():
username = input("Username: ")
password = input("Password: ")
if username == "William" and password == "qwerty123":
print("Welcome!")Створюємо таймер, у якому вказуємо, що функція login() почне виконуватися тільки через 5 секунд (спочатку передаємо інтервал часу, а потім саму функцію):
timer = threading.Timer(5.0, login)Після цього обов’язково треба запустити таймер:
timer.start()Ось повний код:
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()Тепер якщо ви запустите цей код, то функція login() і справді запуститься тільки через 5 секунд.
Перший спосіб створення повторювального таймера
Але чи можна зробити так, щоб таймер змушував функцію виконуватися через кожний інтервал часу? Так, але при розгляді цього питання ми зіштовхуємося з однією особливістю потоків у Python.
Ця особливість полягає у тому, що потік можна запустити лише один раз, а тому при використанні циклу ми отримаємо помилку.
Наприклад, спробуємо цей код:
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()Результат:
RuntimeError: threads can only be started onceДля того, щоб зробити таймер повторювальним, нам варто прислухатися до суті помилки й запустити потік лише один раз. Проте нам потрібно буде створити власний окремий клас, який буде наслідувати об’єкт потоку Thread.
Для вирішення цієї проблеми спочатку нам треба створити власний клас:
class RepeatedTimer(threading.Thread):
def __init__(self, event):
threading.Thread.__init__(self)
self.stopped = eventТут важливо знати дві речі:
Клас
RepeatedTimerнаслідує клас потокуThread.Поле
self.stoppedдорівнює зміннійevent, яка приймає значенняTrueабоFalseі характеризує подію виконання та зупинки таймераRepeatedTimer.
Далі нам потрібно створити метод, який би повторював запуски нашої функції:
def run(self):
while not self.stopped.wait(5.0):
no_user_found(first_names=["Alice", "Peter", "Brian", "William"],
usernames=["@aliceee", "@petermorton", "@brianllars", "@w1lly22"])Тут і знаходиться головна логіка таймера. Як ми знаємо, цикл while виконується до тих пір, поки умова є істинною. Тож при запуску програми атрибут event і, відповідно, поле self.stopped будуть дорівнювати False, а not False дорівнює True.
Наступним кроком запускаємо створений таймер:
stop_flag = threading.Event()
custom_thread = RepeatedTimer(stop_flag)
custom_thread.start()Тут оголошуємо об’єкт stop_flag, який якраз таки і буде тим атрибутом event, який приймає наш клас RepeatedTimer.
Після 10-ти секунд ми змінюємо значення об’єкта stop_flag на True, а так як not True дорівнює False, то цикл while одразу зупиниться.
stop_flag = threading.Event()
custom_thread = RepeatedTimer(stop_flag)
custom_thread.start()
time.sleep(10)
stop_flag.set()Цей код буде працювати рівно 10 секунд, а потім наш об’єкт stop_flag зупинить програму.
Повний код першого способу
Ось повний код першого способу створення повторювального таймера:
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()Другий спосіб створення повторювального таймера
Тим не менш, є й інший спосіб створити повторювальний таймер. Він відрізняється тим, що замість класу потоку Thread наш клас буде наслідувати конкретно клас Timer.
Ось наша функція, створена раніше
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.")Ось код другого варіанту створення повторювального таймера:
class RepeatedTimer(threading.Timer):
def run(self):
while not self.finished.wait(self.interval):
self.function(*self.args, **self.kwargs)Тут логіка роботи така сама, як і у попередньому варіанті, бо self.finished, який стоїть в умові, є об’єктом класу Event(), який ми використовували у минулому прикладі.
Змінна self.interval — це інтервал часу, через який наша функція self.function буде повторюватися.
Тоді все стає зрозумілішим, бо спочатку умова у циклі while дорівнює not False, тобто True. Тому поки умова буде істинною, функція буде повторюватися знову і знову.
Тілом циклу у даному випадку є рядок self.function(*args, **kwargs), який характеризує функцію, яку ми будемо передавати у клас RepeatedTimer.
Тим часом *args та **kwargs — це спеціальні параметри, які допомагають адаптувати цей код до будь-яких функцій, які ми тільки можемо передати у RepeatedTimer, адже ніхто не може передбачити, яку функцію та зі скількома параметрами умовний розробник захоче передати.
Саме завдяки зірочкам (*) та (**) наш повторювальний таймер буде працювати незалежно від кількості параметрів, які ми задамо для функції no_user_found().
Якщо ви не стикалися з параметрами *args та **kwargs до цього, то раджу почитати про це у розділі нижче.
Далі створюємо об’єкт timer, використовуючи наш клас RepeatedTimer, куди ми спочатку передаємо інтервал часу для повторення функції та саму функцію no_user_found.
Варто зауважити, що другий варіант реалізації повторювального таймера виключає запис параметрів у функцію тим чином, яким ми це робили раніше, а саме:
no_user_found(first_names=["Alice", "Peter", "Brian", "William"],
usernames=["@aliceee", "@petermorton", "@brianllars", "@w1lly22"])Замість цього нам потрібно буде вказати значення параметрів у вигляді словника та передати параметру kwargs, який і відповідає за роботу зі словниками:
timer = RepeatedTimer(1, no_user_found, kwargs={
"first_names": ["Alice", "Peter", "Brian", "William"],
"usernames": ["@aliceee", "@petermorton", "@brianllars", "@w1lly22"]
})
timer.start()Тепер коротко розберемо, що ми передали у клас RepeatedTimer():
Спершу передаємо параметр
interval.Далі передаємо саму функцію
no_user_found.Словник із списками.
Зупиняємо таймер (і виконання функції, відповідно) через 10 секунд:
time.sleep(10)
timer.cancel()Також треба розуміти, що наша функція буде повторюватися саме протягом цих 10 секунд, допоки ми не скасуємо таймер.
Повний код другого способу
Ось повний код другого способу створення повторювального таймера:
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()Спеціальні параметри *args та **kwargs
У Python позначення *args — це позиційні аргументи, тобто списки, множини і т. д.
Тут варто розуміти, що важливі тут лише позначення зірочок, а не самі записи args чи kwargs. Саме записи (*) або (**) і відповідають за правильне розкладання елементів представлених у цьому підрозділі типів даних.
Коли ми передаємо такі типи даних у функцію як параметри, то одна зірочка (*) дістає усі елементи у цих типах даних і надає їх функції у вигляді звичайних значень, що записані через кому.
Ось приклад:
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)Результат:
USER INFO
============================
First name: John
Last name: Peterson
Username: @johnnypi
Password: adminЯк бачимо, за допомогою однієї зірочки ми змогли легко розбити усі елементи списку на окремі параметри, які вказали при оголошенні функції.
Цей код за результатом рівноцінний отакому запису:
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])Просто перший запис є набагато зручнішим завдяки чіткому розбиттю елементів списку. Тут же нам довелося записати значення усіх параметрів вручну при виклику функції.
Позначення **kwargs означає іменовані аргументи, тобто словники. В іншому дві зірочки (**) роблять усе те саме, що й одна, але вони працюють з різними типами даних.
Приклад:
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)Цей код рівноцінний такому запису:
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"])Знову-таки, перший варіант з використанням двох зірочок є кращим.
Висновок
У цій статті ми ознайомилися з модулем threading та класом Timer. Також ми детально розібрали два варіанти того, як зробити повторювальний таймер.
