Случались ли у вас ситуации, когда приложение Python должно не сразу выполняться? Обычно требуется, чтобы запуск кода происходил как можно в более сжатые сроки. Тем не менее, иногда лучше дать программе немного «отдохнуть». Сегодня мы поговорим о функции sleep(), которая реализует эту возможность.
Что такое функция sleep()
С помощью функции sleep() разработчик может сделать так, чтобы программа выполняла определенную часть кода не сразу, а по прошествии какого-то времени. В каких ситуациях эта функция может быть полезной? Например, если необходимо дождаться, пока будет приложение скачано, загружено либо появится определенный элемент на экране.
Таким образом, функция sleep() многофункциональная, и может использоваться как для того, чтобы подождать определенное время, так и дождаться до появления какого-то события.
Эта функция пригодится тем разработчикам, которые владеют навыками создания приложений на среднем уровне.
Использование time.sleep() для вызова функции sleep()
Если необходимо осуществить отсрочку выполнения кода на определенное время, необходимо воспользоваться функцией sleep(), которая относится к модулю time.
Давайте приведем пример, как можно воспользоваться данной функцией.
import time time.sleep(3) # Сон в 3 секунды
В Python 3.5 и выше принцип работы time.sleep() немного отличается. Даже если сон прерывается сигналом, все равно отсрочка произойдет на указанное количество секунд. Правда, если такой сигнал оповещает об ошибке, то такого случая это не касается.
Вы можете проверить, насколько долго будет длиться сон, с использованием модуля Python timeit.
$ python3 -m timeit -n 3 "import time; time.sleep(3)" 3 loops, best of 3: 3 sec per loop
А такая программа будет выполнять проверку того, упал ли какой-то из сайтов. А чтобы не создавать излишнюю нагрузку, такая проверка будет выполняться с определенной периодичностью. Чтобы выполнить такой анализ, можно воспользоваться системным вызовом sleep().
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(), принимающего в качестве аргумента адрес сайта. После этого функцией совершается попытка открыть этот адрес сайта с urllib. Если появляется ошибка типа HTTPError, URLError, приложение осуществляет перехват этого исключения и выводит результат на экран. На практике, с большой вероятностью, придется записать данные об этом исключении и отправить соответствующую информацию администратору.
В случае отсутствия исключений, этот код будет без проблем выполняться. Причем нет разницы, что будет происходить, программа будет уходить в сон ровно на 60 секунд. Это означает, что и приложение будет осуществлять попытку подключиться к сайту каждые 60 секунд.
Обратите внимание, что URL, который в предыдущем фрагменте кода используется, с ошибками. Следовательно, вывод командной строки будет таким (обновляться он, как мы поняли, будет каждые 60 секунд).
HTTPError: 404 для http://www.google.com/py
Теперь давайте попробуем внести некоторые изменения в этот фрагмент, используя уже тот URL, который будет 100% гарантированно работать. Программа может быть перезапущена. Проверьте, какой получился результат.
sleep() с декораторами – возможен ли совместный вызов?
В ряде ситуаций необходимо функцию, которая была безуспешно выполнена в первый раз, перезапустить. Нередко это надо делать, когда после чрезмерной нагрузки на сервер необходимо осуществить попытку загрузки файла. Обычно никто не желает производить частые вызовы данных с сервера. Следовательно, использование sleep() между запросами – очень хорошая идея.
Помимо этого, функция sleep() может использоваться в тех ситуациях, когда нужно проверить состояние интерфейса в ходе тестирования. От ПК, на каком производится запуск теста, зависит скорость, с которой будет осуществляться загрузка пользовательского интерфейса. Она может происходить как быстрее, так и не так быстро. Таким образом, содержимое, которое будет показываться на мониторе в ходе проверки приложением чего-то также изменяется.
Как решить эту проблему? Например, указать приложению, чтобы оно приостановило работу, а потом выполнить проверку через несколько секунд.
Чтобы добавить системный вызов sleep() в Python, возможно использование декоратора как между прохождением, так и недачей теста.
Давайте рассмотрим такой пример.
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, что по умолчанию равняется трем. Также внутри функции sleep() есть другая, the_real_decorator(), принимающая декорируемую функцию.
В результате, функцией, которая является самой внутренней, wrapper(), принимаются аргументы, а также ключевые слова, которые передаются декорируемой функции. Здесь все и происходит. С помощью цикла while мы повторно вызываем функцию. При возникновении исключения, происходит вызов time.sleep(), увеличивается счетчик попыток retries и осуществляется попытка запуска функции.
Помимо этого, производится переписывание uptime_bot() для нового декоратора.
@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(). Помимо этого, был удален оригинальный while, а также старый вызов sleep(60). Декоратор сделал все для этого.
Также мы добавили raise внутри блоков, которые выполняют обработку ошибок. Это необходимо, чтобы декоратор правильно выполнял свои функции. Помимо этого, можно написать такой декоратор, который будет отвечать за ошибки. Тем не менее, в нашем случае лучше оставить его в неизменном виде, поскольку ошибки касаются исключительно urllib. В этом случае набор функций, которые он будет задействовать, будет значительно шире.
Sleep() в потоках
В некоторых случаях может понадобиться использование sleep() для потока. Например, осуществить запуск скрипта миграции для базы данных, в которой хранится огромное число записей. Здесь критически важно не допустить простоя, а также чтобы не было надобности ждать на протяжении большего временного промежутка, чем необходимо для завершения миграции. Для этого можно использовать потоки.
Под потоками подразумевается один из методов, с помощью которого становится возможным обеспечение конкурентности в Python. Пользователь может осуществлять запуск сразу нескольких потоков. Таким образом можно сделать программу более производительной.
Чтобы для клиентов приложение работало идеально по части скорости, каждый из потоков должен функционировать на протяжении небольшого временного отрезка, после чего уходить в сон. Это делается двумя способами:
- Применить time.sleep(), как мы показывали ранее. Ниже приведем фрагмент кода, который демонстрирует то, как это работает.
- Воспользоваться Event.wait(), расположенном в модуле threading.
time.sleep()
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()
Вот пример кода, демонстрирующего, как используется time.sleep() в threading.
Event.wait()
Преимущество этого метода в его отзывчивости. Вот, как он работает.
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()
