Знакома ли вам ситуация, когда программа Python должна выполняться не сразу? В большинстве случаев требуется, чтобы код запускался как можно скорее. Однако порой перед работой оптимальнее будет дать программе немного поспать.
Содержание статьи
- Вызов sleep() через time.sleep()
- Вызов sleep() с декораторами
- Вызов sleep() в потоках
- Использование time.sleep() в threading
- Использование Event.wait() в многопоточности Python
- Вызов sleep() в Async IO
- Вызов sleep() в Tkinter и wxPython
- Метод after() — Погружение в сон для Tkinter
- Метод CallLater() — Погружение в сон для wxPython Python
В Python есть возможность вызвать функцию sleep() для симуляции задержки в выполнении программы. Быть может, вам нужно дождаться загрузки, скачивания или появления графического объекта на экране. Также может потребоваться сделать паузу между вызовами к веб API или запросами к базе данных. В таких случаях поможет добавление вызова функции sleep() в программу.
Главные аспекты данного руководства по вызову sleep() в Python:
time.sleep();- Декораторы;
- Потоки;
- Async IO;
- Графический пользовательский интерфейс GUI.
Данная статья предназначена для разработчиков Python среднего уровня, что стремятся повысить свою квалификацию. Если это похоже на вас, приступим!
Вызов sleep() через time.sleep()
В Python есть встроенная поддержка для погружения программы в сон. У модуля time есть функция sleep(), что позволяет отсрочить выполнение вызываемого потока на указанное количество секунд.
Далее дан пример использования time.sleep():
|
1 2 |
import time time.sleep(3) # Сон в 3 секунды |
При запуске кода из консоли, задержку нужно проводить перед вводом нового оператора в REPL.
На заметку: В Python 3.5 разработчики слегка изменили поведение
time.sleep(). Благодаря новой системе вызоваsleep()эффект отсрочки будет длиться как минимум на продолжении указанного количества секунд, даже в том случае, если сон прерывается сигналом. Однако, это не касается случаев, если сигнал является признаком вызова исключения.
Вы можете протестировать, как долго продлиться сон с помощью модуля Python timeit:
|
1 2 |
$ python3 -m timeit -n 3 "import time; time.sleep(3)" 3 loops, best of 3: 3 sec per loop |
Здесь модуль timeit запускается с параметром -n, что указывает timeit, сколько раз выполнять последующий оператор. Можно заметить, что timeit выполнил оператор 3 раза, а лучшее время длилось 3 секунды, чего и следовало ожидать.
По умолчанию timeit будет запускать код миллион раз. Если бы вы запустили вышеуказанный код, оставив значение -n по умолчанию, тогда при 3 секундах на итерацию код завис бы примерно на 34 дня! У модуля timeit есть несколько других настроек для командной строки, с которыми можно ознакомиться в документации.
Создадим что-то более практичное. Системному администратору всегда нужно быть в курсе, если какой-то из сайтов упал. Вы бы хотели иметь возможность проверить код состояния сайта регулярно, но запрашивать веб сервер постоянно нельзя, ведь это сильно повлияет на производительность. В Python одним из простых способов совершить такую проверку является использование системного вызова sleep():
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
import time import urllib.request import urllib.error def uptime_bot(url): while True: try: conn = urllib.request.urlopen(url) except urllib.error.HTTPError as e: # Отправка admin / log print(f'HTTPError: {e.code} для {url}') except urllib.error.URLError as e: # Отправка admin / log print(f'URLError: {e.code} для {url}') else: # Сайт поднят print(f'{url} поднят') time.sleep(60) if __name__ == '__main__': url = 'http://www.google.com/py' uptime_bot(url) |
Здесь создается uptime_bot(), что принимает URL в качестве аргумента. Затем функция пытается открыть данный URL c urllib. При возникновении HTTPError или URLError программа перехватывает ошибку и выводит на экран. На практике вам, скорее всего, придется зафиксировать ошибку и отправить письмо веб-мастеру или системному администратору.
Если ошибок нет, код спокойно выполняется. Вне зависимости от того, что произойдет, программа уходит в сон на 60 секунд. Это значит, что доступ к сайту будет раз за минуту. URL, используемый в примере, содержит ошибки. Ежеминутный вывод на консоли выглядит следующим образом:
|
1 |
HTTPError: 404 для http://www.google.com/py |
Попробуйте обновить код, используя проверенный хороший URL, к примеру https://www.google.com/. После этого вы можете перезапустить программу и проверить, что изменилось. Также можно попробовать обновить код для отправки сообщения или записи об ошибке. Для получения более подробной информации можете ознакомиться со статьями отправка писем smtp и логирование.
Вызов sleep() с декораторами
В некоторых случаях нужно повторно запустить неудачно выполненную в первый раз функцию. Зачастую это происходит, когда требуется повторить загрузку файла ввиду ранней перегрузки сервера. Как правило, никто не хочет делать частые запросы на серверы, поэтому добавление в Python вызова sleep() между каждым запросом предпочтительно.
Другим возможным случаем использования sleep() является необходимость проверки состояния пользовательского интерфейса во время автоматического теста. В зависимости от компьютера, на котором запускается тест, пользовательский интерфейс может грузиться быстрее или медленнее обычного. Это может изменить отображаемое на экране во время проверки программой чего-то.
В данном случае можно указать программе, чтобы та погрузилась в сон на мгновенье и затем проверить все опять через несколько секунд. Это может означать разницу между прохождением или провалом теста.
Для добавления системного вызова sleep() в Python можно использовать декоратор в каждом из данных случаев. Разберем следующий пример:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
import time import urllib.request import urllib.error def sleep(timeout, retry=3): def the_real_decorator(function): def wrapper(*args, **kwargs): retries = 0 while retries < retry: try: value = function(*args, **kwargs) if value is None: return except: print(f'Сон на {timeout} секунд') time.sleep(timeout) retries += 1 return wrapper return the_real_decorator |
sleep() является вашим декоратором. Он принимает значение timeout и количество раз для повтора retry, что по умолчанию равняется 3. Внутри sleep() есть другая функция, the_real_decorator(), которая принимает декорируемую функцию.
В конечном итоге самая внутренняя функция wrapper() принимает аргументы и ключевые слова, которые вы передаете декорируемой функции. Здесь все и происходит! Используется цикл while, чтобы повторить вызов функции. Если возникла ошибка, вызывается time.sleep(), увеличивается счетчик попыток retries и повторяется попытка запуска функции.
Теперь переписывается uptime_bot() для использования нового декоратора:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
@sleep(3) def uptime_bot(url): try: conn = urllib.request.urlopen(url) except urllib.error.HTTPError as e: # Отправка admin / log print(f'HTTPError: {e.code} для {url}') # Повторное поднятие ошибки исключения для декоратора raise urllib.error.HTTPError except urllib.error.URLError as e: # Отправка admin / log print(f'URLError: {e.code} для {url}') # Повторное поднятие ошибки исключения для декоратора raise urllib.error.URLError else: # Сайт поднят print(f'{url} поднят') if __name__ == '__main__': url = 'http://www.google.com/py' uptime_bot(url) |
Здесь вы декорируете uptime_bot() с помощью sleep() в 3 секунды. Вы также удалили оригинальный цикл while и старый вызов sleep(60). Декоратор теперь позаботится об этом.
Другое изменение состоит в добавлении raise внутри блоков, отвечающих за обработку исключений. Это нужно для правильной работы декоратора. Можно также написать декоратор, чтобы он отвечал за ошибки, однако ввиду того, что исключения касаются только urllib, может быть лучше сохранить декоратор в текущем состоянии. В таком случае он будет работать c более широким ассортиментом функций.
На заметку: При желании более подробно узнать о том, как справляться с исключениями в Python, можете ознакомиться со статьей: Обработка исключений в Python
Декоратору можно добавить несколько улучшений. Если число попыток заканчивается, и он по-прежнему проваливается, тогда можно сделать так, чтобы он повторно вызвал последнюю ошибку. Декоратор подождет 3 секунды после последней неудачи, что не всегда нужно. Можете попробовать поэкспериментировать самостоятельно.
Вызов sleep() в потоках
Могут возникнуть ситуации, когда в Python требуется добавить вызов sleep() для потока. К примеру, запуск скрипта миграции для базы данных с миллионами записей. Здесь важно избежать простоя, а также не ждать дольше необходимого для завершения миграции, поэтому можно использовать потоки.
На заметку: Потоки являются одним из методов использования конкурентности в Python. Можно запустить несколько потоков одновременно, чтобы увеличить производительность приложения. Если потоки в Python являются для вас новой темой, ознакомьтесь со статьей модуль threading.
Чтобы клиенты не замечали какого-либо замедления, каждый поток должен работать в течение короткого периода времени, а затем уходить в сон. Есть два способа сделать это:
- Использовать
time.sleep()как ранее; - Использовать
Event.wait()из модуляthreading;
Начнем с разбора time.sleep().
Использование time.sleep() в threading
Python Logging Cookbook является хорошим примером использования time.sleep(). Модуль логирования logging является потоко-безопасным, поэтому в данном примере он будет полезнее, чем операторы print(). В основе следующего кода лежит данный пример:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 |
import logging import threading import time def worker(arg): while not arg["stop"]: logging.debug("рабочий поток вносится") time.sleep(1) def main(): logging.basicConfig( level=logging.DEBUG, format="%(relativeCreated)6d %(threadName)s %(message)s" ) info = {"stop": False} thread = threading.Thread(target=worker, args=(info,)) thread_two = threading.Thread(target=worker, args=(info,)) thread.start() thread_two.start() while True: try: logging.debug("Добавление из главного потока") time.sleep(0.75) except KeyboardInterrupt: info["stop"] = True logging.debug('Остановка') break thread.join() thread_two.join() if __name__ == "__main__": main() |
Здесь для создания двух потоков используется модуль Python threading. Также создается объект входа, что будет вводить threadName в stdout. Затем начинаются оба потока и инициируется цикл для каждого входа из главного потока. Для фиксирования пользователя используется KeyboardInterrupt при нажатии ^Ctrl+C.
Попробуйте запустить вышеуказанный код в терминале. Ваш вывод должен походить на следующий:
|
1 2 3 4 5 6 7 8 9 10 11 12 |
2 Thread-1 рабочий поток вносится 4 Thread-2 рабочий поток вносится 4 MainThread Добавление из главного потока 755 MainThread Добавление из главного потока 1004 Thread-1 рабочий поток вносится 1006 Thread-2 рабочий поток вносится 1506 MainThread Добавление из главного потока 2005 Thread-1 рабочий поток вносится 2007 Thread-2 рабочий поток вносится 2257 MainThread Добавление из главного потока 3007 Thread-1 рабочий поток вносится 3008 MainThread Добавление из главного потока |
Когда каждый поток работает, а затем уходит в сон, выходные данные выводятся в консоль. Теперь, разобрав пример, вы сможете использовать данные концепции в своем собственном коде.
Использование Event.wait() в многопоточности Python
Модуль threading предоставляет Event(), которого можно использовать как time.sleep(). Однако преимущество Event() в том, что он более отзывчив. Причина в том, что когда событие установлено, программа сразу выходит из цикла. В Python с time.sleep() коду надо будет подождать завершения вызова sleep() до выхода из потока.
Причина, по которой здесь лучше использовать wait() в том, что он не блокируется, в то время, как time.sleep() блокируется. Это значит, что при использовании time.sleep() вы заблокируете выполнение основного потока, пока тот будет ждать завершения вызова sleep(). wait() решает данную проблему. Более подробнее прочитать о принципах работы потоков можно в документации.
Далее показан пример добавления в Python вызова sleep() с Event.wait():
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
import logging import threading def worker(event): while not event.isSet(): logging.debug("рабочий поток вносится") event.wait(1) def main(): logging.basicConfig( level=logging.DEBUG, format="%(relativeCreated)6d %(threadName)s %(message)s" ) event = threading.Event() thread = threading.Thread(target=worker, args=(event,)) thread_two = threading.Thread(target=worker, args=(event,)) thread.start() thread_two.start() while not event.isSet(): try: logging.debug("Добавление из главного потока") event.wait(0.75) except KeyboardInterrupt: event.set() break if __name__ == "__main__": main() |
В данном примере создается threading.Event() и передается к worker(). Вспомните, что в предыдущем примере вместо этого передавался словарь.
Затем устанавливаются циклы для проверки, настроено ли событие event. Если это не так, тогда код выведет сообщение и немного подождет перед повторной проверкой. Для установки события можно воспользоваться комбинацией ^Ctrl+C. Как только событие установлено, worker() вернется, и цикл оборвется, завершив программу.
На заметку: При желании подробнее узнать о словарях в Python можете ознакомиться со статьей по работе со словарями в Python на примерах.
Рассмотрите подробнее код выше. Как бы вы передали разное время сна каждому работающему потоку? Справитесь с задачей? Не бойтесь экспериментировать!
Вызов sleep() с Async IO на примерах
Асинхронные возможности были добавлены в Python 3.4, и с тех пор данный аспект постоянно распространяется и улучшается. Асинхронное программирование является типом параллельного программирования, что позволяет одновременно запускать множество задач. По завершении задачи выводится уведомления для основного потока.
Модуль asyncio позволяет добавлять в Python вызов sleep() асинхронно.
Вот пример, данный в официальной документации Python:
|
1 2 3 4 5 6 7 8 9 |
import asyncio async def main(): print('Hello ...') await asyncio.sleep(1) print('... World!') # Python 3.7+ asyncio.run(main()) |
В данном примере запускается main(), что погружается в сон на секунду между двумя вызовами print().
Вот более подробный пример из раздела о Сопрограммах и задачах документации asyncio:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
import asyncio import time async def output(sleep, text): await asyncio.sleep(sleep) print(text) async def main(): print(f"Started: {time.strftime('%X')}") await output(1, 'First') await output(2, 'Second') await output(3, 'Third') print(f"Ended: {time.strftime('%X')}") # Python 3.7+ asyncio.run(main()) |
В данном коде создается рабочий поток output(), что принимает количество секунд для sleep и выводит text. После этого используется ключевое слово Python await для ожидания запуска кода output(). Здесь требуется await, так как output() был отмечен как функция async, и теперь не может вызываться как обычная функция.
При запуске кода программа выполнит await 3 раза. Код будет ждать 1, 2 и 3 секунды, общее время ожидания равно 6 секундам. Можно также переписать код таким образом, чтобы задачи выполнялись параллельно:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
import asyncio import time async def output(text, sleep): while sleep > 0: await asyncio.sleep(1) print(f'{text} counter: {sleep} seconds') sleep -= 1 async def main(): task_1 = asyncio.create_task(output('First', 1)) task_2 = asyncio.create_task(output('Second', 2)) task_3 = asyncio.create_task(output('Third', 3)) print(f"Started: {time.strftime('%X')}") await task_1 await task_2 await task_3 print(f"Ended: {time.strftime('%X')}") if __name__ == '__main__': asyncio.run(main()) |
Теперь вы используете концепт задач, что можно создать через create_task(). При использовании задач в asyncio Python будет запускать задачи асинхронно. Таким образом, выполнение программы завершится через 3 секунды вместо 6.
Вызов sleep() в Tkinter и wxPython
Вызовы sleep() в Python можно добавить не только для приложений командной строки. При создании графического пользовательского интерфейса (GUI) периодически нужно добавлять отсрочки. К примеру, при создании приложения FTP для скачивания около миллиона файлов будет разумно добавить вызов sleep() между партиями, чтобы снизить нагрузку на сервер.
GUI код выполнит всю обработку в основном потоке, называемом циклом обработки событий, или event loop. При использовании time.sleep() внутри кода GUI заблокируется цикл обработки событий.
Со стороны пользователя приложение может зависнуть. Пользователь не сможет взаимодействовать с вашим приложением, пока оно погружено в сон вследствие работы данного метода. В Windows можно даже получить предупреждение о том, что ваше приложение теперь не отвечает.
К счастью, помимо time.sleep(), можно использовать некоторые другие методы специально для этой задачи. Далее мы рассмотрим, как добавить вызовы sleep() в Tkinter и wxPython.
Метод after() — Погружение в сон для Tkinter
tkinter является частью стандартной библиотеки Python. В случае, если вы используете заранее установленную версию Python на Linux или Mac, он может быть вам недоступен. При получении ошибки ImportError стоит самостоятельно добавить его в систему. В том случае, если вы ранее установили Python сами, tkinter должен быть доступен.
Начнем с разбора примера, где используется time.sleep(). Запустите следующий код и посмотрите, что произойдет при неправильном добавлении вызова sleep() в Python:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
import tkinter import time class MyApp: def __init__(self, parent): self.root = parent self.root.geometry("400x400") self.frame = tkinter.Frame(parent) self.frame.pack() b = tkinter.Button(text="click me", command=self.delayed) b.pack() def delayed(self): time.sleep(3) if __name__ == "__main__": root = tkinter.Tk() app = MyApp(root) root.mainloop() |
После запуска кода нажмите кнопку в GUI. Кнопка не будет реагировать три секунды, ожидая завершения sleep(). Если в приложении есть другие кнопки, на них тоже нельзя будет нажать. Закрыть приложение во время сна нельзя, так как оно не будет откликаться на событие закрытия.
Для должного погружения tkinter в сон потребуется использовать after():
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
import tkinter class MyApp: def __init__(self, parent): self.root = parent self.root.geometry("400x400") self.frame = tkinter.Frame(parent) self.frame.pack() self.root.after(3000, self.delayed) def delayed(self): print('Я задержался') if __name__ == "__main__": root = tkinter.Tk() app = MyApp(root) root.mainloop() |
Здесь создается приложение, высота которого 400 пикселей, и ширина также 400 пикселей. На нем нет виджетов. Оно только показывает фрейм. Затем вызывается self.root.after(), где self.root является отсылкой к объекту Tk(). after() принимает два аргумента:
- Количество миллисекунд для сна;
- Метод который вызовется после завершения сна.
В данном случае приложение выведет строку в стандартный поток вывода (stdout) через 3 секунды. Можно рассматривать after() как Tkinter-версию того же time.sleep(), только он добавляет способность вызова функции после завершения сна.
Данную функциональность можно использовать для улучшения работы пользователя. Добавив в Python вызов sleep(), можно ускорить процесс загрузки приложения, после чего начать какой-то длительный процесс. В таком случае пользователю не придется ждать открытия приложения.
Метод CallLater() — Погружение в сон для wxPython Python
Между wxPython и Tkinter есть два важных различия:
- В wxPython намного больше виджетов;
- wxPython стремится выглядеть нативным на любой платформе.
Фреймворк wxPython не поставляется с Python вместе, поэтому его нужно установить wxPython самостоятельно. Если вы не знакомы с wxPython, можете изучить курс из 55 уроков по wxPython.
Для добавления вызова sleep() в wxPython можно использовать wx.CallLater():
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
import wx class MyFrame(wx.Frame): def __init__(self): super().__init__(parent=None, title='Привет, мир!') wx.CallLater(4000, self.delayed) self.Show() def delayed(self): print('Я задержался') if __name__ == '__main__': app = wx.App() frame = MyFrame() app.MainLoop() |
Здесь напрямую создается подкласс wx.Frame и затем вызывается wx.CallLater(). Данная функция принимает такие же параметры, что и after() в Tkinter:
- Количество миллисекунд для сна;
- Метод который вызовется сразу после завершения сна.
При запуске данного кода появится небольшое пустое окно без виджетов. Через 4 секунды в стандартном потоке вывода (stdout) появится строка 'Я задержался'.
Одним из преимуществ wx.CallLater() является его поточная безопасность. Данный метод можно использовать внутри потока для вызова функции, что в основном приложении wxPython.
Заключение
В данном руководстве вы познакомились с новой полезной техникой для работы в Python. Теперь вы знаете, как добавить задержку для ускорения работы приложений и предотвращения использование ими системных ресурсов. Вы также можете использовать вызовы Python sleep(), чтобы помочь графическому интерфейсу GUI работать более эффективно. Все это может усовершенствовать опыт пользователя при работе в приложении.
Подведем итоги. Основные, рассмотренные в статье инструменты, для добавления вызовов sleep() в Python:
time.sleep();- Декораторы;
- Потоки;
asyncio;- Tkinter;
- wxPython;
Теперь вы можете использовать полученные знания и погружать ваш код на Python в сон.
Являюсь администратором нескольких порталов по обучению языков программирования Python, Golang и Kotlin. В составе небольшой команды единомышленников, мы занимаемся популяризацией языков программирования на русскоязычную аудиторию. Большая часть статей была адаптирована нами на русский язык и распространяется бесплатно.
E-mail: vasile.buldumac@ati.utm.md
Образование
Universitatea Tehnică a Moldovei (utm.md)
- 2014 — 2018 Технический Университет Молдовы, ИТ-Инженер. Тема дипломной работы «Автоматизация покупки и продажи криптовалюты используя технический анализ»
- 2018 — 2020 Технический Университет Молдовы, Магистр, Магистерская диссертация «Идентификация человека в киберпространстве по фотографии лица»