Python sleep(): как добавить задержки в свой код

Оглавление

• Добавление вызова Python sleep() с помощью time.sleep()
• Добавление вызова Python sleep() С помощью декораторов
• Добавление вызова Python sleep() с потоками
  • Используя время сна.()
  • Используя Event.wait, дождитесь()
• Добавление вызова Python sleep() С асинхронным вводом-выводом
• Добавление вызова Python sleep() с графическим интерфейсом
  • Сплю в Tkinter
  • Спящий в wxPython
• Заключение

Приходилось ли вам когда-нибудь заставлять вашу программу на Python чего-либо ждать? В большинстве случаев вы хотели бы, чтобы ваш код выполнялся как можно быстрее. Но бывают случаи, когда в ваших интересах дать вашему коду какое-то время поспать.

Например, вы можете использовать вызов Python sleep() для имитации задержки в вашей программе. Возможно, вам нужно дождаться загрузки файла или графического изображения, которое будет загружено или выведено на экран. Возможно, вам даже потребуется сделать паузу между вызовами веб-API или между запросами к базе данных. Добавление Python sleep() обращения к вашей программе могут помочь в каждом из этих случаев и во многих других!

В этом руководстве вы узнаете, как добавить вызовы Python sleep() с помощью:

• time.sleep()
• Декораторы
• Темы
• Асинхронный ввод-вывод
• Графические интерфейсы пользователя

Эта статья предназначена для разработчиков среднего уровня, которые хотят расширить свои знания Python. Если это похоже на вас, то давайте начнем!

Добавление вызова Python sleep() С помощью time.sleep()

В Python есть встроенная поддержка перевода вашей программы в спящий режим. В модуле time есть функция sleep(), с помощью которой вы можете приостановить выполнение вызывающего потока на любое количество секунд, которое вы укажете.

Вот пример того, как использовать time.sleep():

>>> import time
>>> time.sleep(3) # Sleep for 3 seconds

Если вы запустите этот код в своей консоли, то у вас должна возникнуть задержка, прежде чем вы сможете ввести новую инструкцию в ОТВЕТ.

Примечание: В Python 3.5 разработчики ядра немного изменили поведение time.sleep(). Новый системный вызов Python sleep() будет длиться не менее указанного вами количества секунд, даже если режим ожидания будет прерван сигналом. Однако это не применяется, если сам сигнал вызывает исключение.

Вы можете проверить, как долго длится режим ожидания, используя модуль Python timeit:

$ python3 -m timeit -n 3 "import time; time.sleep(3)"
3 loops, best of 5: 3 sec per loop

Здесь вы запускаете модуль timeit с параметром -n, который указывает timeit, сколько раз следует выполнить следующую инструкцию. Вы можете видеть, что timeit запустил инструкцию 3 раза и что наилучшее время выполнения составило 3 секунды, что и ожидалось.

По умолчанию количество раз, которое timeit выполнит ваш код, равно миллиону. Если бы вы запустили приведенный выше код с параметром по умолчанию -n, то при 3 секундах на итерацию ваш терминал зависал бы примерно на 34 дня! Модуль timeit имеет несколько других опций командной строки, с которыми вы можете ознакомиться в его документации .

Давайте создадим что-то более реалистичное. Системный администратор должен знать, когда один из его веб-сайтов выходит из строя. Вы хотите иметь возможность регулярно проверять код состояния веб-сайта, но вы не можете постоянно запрашивать веб-сервер, иначе это повлияет на производительность. Один из способов выполнить эту проверку - использовать системный вызов Python 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:
            # Email admin / log
            print(f'HTTPError: {e.code} for {url}')
        except urllib.error.URLError as e:
            # Email admin / log
            print(f'URLError: {e.code} for {url}')
        else:
            # Website is up
            print(f'{url} is up')
        time.sleep(60)

if __name__ == '__main__':
    url = 'http://www.google.com/py'
    uptime_bot(url)

Здесь вы создаете uptime_bot(), который принимает URL в качестве аргумента. Затем функция пытается открыть этот URL с помощью urllib. Если есть HTTPError или URLError, то программа перехватывает это и выводит сообщение об ошибке. (В реальной среде вы бы зарегистрировали ошибку и, вероятно, отправили электронное письмо веб-мастеру или системному администратору.)

Если ошибок не возникает, то ваш код выводит сообщение о том, что все в порядке. Независимо от того, что произойдет, ваша программа будет находиться в спящем режиме в течение 60 секунд. Это означает, что вы заходите на веб-сайт только раз в минуту. URL-адрес, используемый в этом примере, неверен, поэтому он будет выводить на вашу консоль следующее сообщение раз в минуту:

HTTPError: 404 for http://www.google.com/py

Обновите код, чтобы использовать заведомо правильный URL, например http://www.google.com. Затем вы можете повторно запустить его, чтобы убедиться, что он успешно работает. Вы также можете попробовать обновить код, чтобы отправить электронное письмо или зарегистрировать ошибки. Для получения дополнительной информации о том, как это сделать, ознакомьтесь с разделами Отправка электронных писем с помощью Python и Регистрация в Python.

Добавление вызова Python sleep() С помощью декораторов

Бывают случаи, когда вам нужно повторить попытку выполнения функции, которая завершилась сбоем. Один из популярных вариантов использования этого метода - когда вам нужно повторить попытку загрузки файла, потому что сервер был занят. Обычно вам не захочется делать запрос к серверу слишком часто, поэтому желательно добавлять вызов Python sleep() между каждым запросом.

Другой пример использования, с которым я лично сталкивался, - это когда мне нужно проверить состояние пользовательского интерфейса во время автоматического тестирования. Пользовательский интерфейс может загружаться быстрее или медленнее, чем обычно, в зависимости от компьютера, на котором я запускаю тест. Это может изменить то, что отображается на экране в тот момент, когда моя программа что-то проверяет.

В этом случае я могу перевести программу в спящий режим на мгновение, а затем перепроверить все через секунду или две. Это может означать разницу между успешным и неудачным тестом.

Вы можете использовать декоратор, чтобы добавить системный вызов Python 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'Sleeping for {timeout} seconds')
                    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(), чтобы использовать ваш новый декоратор:

@sleep(3)
def uptime_bot(url):
    try:
        conn = urllib.request.urlopen(url)
    except urllib.error.HTTPError as e:
        # Email admin / log
        print(f'HTTPError: {e.code} for {url}')
        # Re-raise the exception for the decorator
        raise urllib.error.HTTPError
    except urllib.error.URLError as e:
        # Email admin / log
        print(f'URLError: {e.code} for {url}')
        # Re-raise the exception for the decorator
        raise urllib.error.URLError
    else:
        # Website is up
        print(f'{url} is up')

if __name__ == '__main__':
    url = 'http://www.google.com/py'
    uptime_bot(url)

Здесь вы добавляете к uptime_bot() значение sleep() продолжительностью в 3 секунды. Вы также удалили исходный while цикл, а также старый вызов sleep(60). Теперь об этом позаботится декоратор.

Еще одно изменение, которое вы внесли, - это добавление raise внутри блоков обработки исключений. Это делается для того, чтобы декоратор работал должным образом. Вы могли бы написать декоратор для обработки этих ошибок, но поскольку эти исключения применимы только к urllib, возможно, вам лучше оставить декоратор таким, какой он есть. Таким образом, он будет работать с более широким спектром функций.

Примечание: Если вы хотите ознакомиться с обработкой исключений в Python, ознакомьтесь с Исключениями Python: введение.

Есть несколько улучшений, которые вы могли бы внести в свой декоратор. Если у него закончатся повторные попытки, но он по-прежнему не будет работать, вы можете попросить его повторно вызвать последнюю ошибку. Декоратор также подождет 3 секунды после последнего сбоя, чего вы, возможно, не хотели бы. Не стесняйтесь использовать это в качестве упражнения!

Добавление вызова Python sleep() С потоками

Бывают также случаи, когда вам может потребоваться добавить вызов Python sleep() в поток. Возможно, вы запускаете сценарий миграции для базы данных с миллионами записей в рабочей среде. Вы не хотите вызывать простои, но и не хотите ждать завершения миграции дольше, чем необходимо, поэтому вы решили использовать потоки.

Примечание: Потоки - это метод обеспечения параллелизма в Python. Вы можете запускать несколько потоков одновременно, чтобы увеличить пропускную способность вашего приложения. Если вы не знакомы с потоками в Python, ознакомьтесь с Введением к потоковой обработке в Python.

Чтобы пользователи не заметили какого-либо замедления, каждый поток должен выполняться в течение короткого периода времени, а затем переходить в режим ожидания. Есть два способа сделать это:

1. Используйте time.sleep() как и раньше.
2. Используйте Event.wait() из модуля threading.

Давайте начнем с рассмотрения time.sleep().

Используя time.sleep()

В Python Logging Cookbook показан хороший пример, в котором используется time.sleep(). Модуль Python logging является потокобезопасным, поэтому для этого упражнения он немного полезнее, чем print() инструкции. Следующий код основан на этом примере:

import logging
import threading
import time

def worker(arg):
    while not arg["stop"]:
        logging.debug("worker thread checking in")
        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("Checking in from main thread")
            time.sleep(0.75)
        except KeyboardInterrupt:
            info["stop"] = True
            logging.debug('Stopping')
            break
    thread.join()
    thread_two.join()

if __name__ == "__main__":
    main()

Здесь вы используете модуль Python threading для создания двух потоков. Вы также создаете объект ведения журнала, который будет записывать threadName в стандартный вывод. Затем вы запускаете оба потока и время от времени запускаете цикл для входа в систему из основного потока. Вы используете KeyboardInterrupt, чтобы перехватить нажатие пользователем Ctrl+C.

Попробуйте запустить приведенный выше код в своем терминале. Вы должны увидеть результат, похожий на следующий:

 0 Thread-1 worker thread checking in
 1 Thread-2 worker thread checking in
 1 MainThread Checking in from main thread
752 MainThread Checking in from main thread
1001 Thread-1 worker thread checking in
1001 Thread-2 worker thread checking in
1502 MainThread Checking in from main thread
2003 Thread-1 worker thread checking in
2003 Thread-2 worker thread checking in
2253 MainThread Checking in from main thread
3005 Thread-1 worker thread checking in
3005 MainThread Checking in from main thread
3005 Thread-2 worker thread checking in

Когда каждый поток запускается, а затем переходит в режим ожидания, выходные данные журнала выводятся на консоль. Теперь, когда вы попробовали пример, вы сможете использовать эти концепции в своем собственном коде.

Используя Event.wait()

Модуль threading предоставляет Event(), который вы можете использовать как time.sleep(). Однако, Event() обладает дополнительным преимуществом - он более отзывчив. Причина этого заключается в том, что при установлении события программа немедленно выходит из цикла. С time.sleep() вашему коду нужно будет дождаться завершения вызова Python sleep(), прежде чем поток сможет завершить работу.

Причина, по которой вы хотели бы использовать wait() здесь, заключается в том, что wait() является неблокирующим , тогда как time.sleep() является блокировка. Это означает, что при использовании time.sleep() вы блокируете продолжение выполнения основного потока, пока он ожидает завершения вызова sleep(). wait() решает эту проблему. Вы можете прочитать больше о том, как все это работает, в документации по потоковой обработке на Python .

Вот как вы добавляете вызов Python sleep() с помощью Event.wait():

import logging
import threading

def worker(event):
    while not event.isSet():
        logging.debug("worker thread checking in")
        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("Checking in from main thread")
            event.wait(0.75)
        except KeyboardInterrupt:
            event.set()
            break

if __name__ == "__main__":
    main()

В этом примере вы создаете threading.Event() и передаете его в worker(). (Напомним, что в предыдущем примере вы вместо этого передали словарь.) Затем вы настраиваете свои циклы, чтобы проверить, установлен ли event. Если это не так, то ваш код выводит сообщение и немного ждет, прежде чем снова проверять. Чтобы задать событие, вы можете нажать Ctrl+C. Как только событие будет установлено, worker() вернется, и цикл прервется, завершая программу.

Примечание: Если вы хотите узнать больше о словарях, ознакомьтесь с Словарями на Python.

Взгляните внимательнее на приведенный выше блок кода. Как бы вы определили разное время ожидания для каждого рабочего потока? Можете ли вы разобраться в этом? Не стесняйтесь выполнять это упражнение самостоятельно!

Добавление вызова Python sleep() С асинхронным вводом-выводом

Асинхронные возможности были добавлены в Python в версии 3.4, и с тех пор этот набор функций постоянно расширяется. Асинхронное программирование - это это разновидность параллельного программирования, которая позволяет выполнять несколько задач одновременно. Когда задача завершается, она уведомляет об этом основной поток.

asyncio это модуль, который позволяет вам добавлять асинхронный вызов Python sleep(). Если вы не знакомы с реализацией асинхронного программирования в Python, ознакомьтесь с Асинхронный ввод-вывод в Python: полное пошаговое руководство и Параллелизм и программирование на Python.

Вот пример из собственной документации Python :

import asyncio

async def main():
    print('Hello ...')
    await asyncio.sleep(1)
    print('... World!')

# Python 3.7+
asyncio.run(main())

В этом примере вы запускаете main() и переводите его в режим ожидания на одну секунду между двумя вызовами print().

Вот более наглядный пример из раздела Сопрограммы и задачи документации asyncio:

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. Затем вы используете ключевое слово await в Python, чтобы дождаться запуска кода output(). Здесь требуется await, потому что output() помечена как функция async, так что вы не можете вызвать его так, как вызвали бы обычную функцию.

Когда вы запустите этот код, ваша программа выполнит await 3 раза. Код будет ждать 1, 2 и 3 секунды, общее время ожидания составит 6 секунд. Вы также можете переписать код таким образом, чтобы задачи выполнялись параллельно:

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.

Добавление вызова Python sleep() С графическим интерфейсом

Приложения командной строки - не единственное место, где вам может понадобиться добавить вызовы Python sleep(). Когда вы создаете Графический интерфейс пользователя (GUI), вам иногда нужно будет добавлять задержки. Например, вы можете создать FTP-приложение для загрузки миллионов файлов, но вам нужно добавить sleep() вызов между пакетами, чтобы не перегружать сервер.

Графический код будет выполнять всю свою обработку и отрисовку в основном потоке, называемом цикл обработки событий. Если вы используете time.sleep() внутри графического интерфейса пользователя, вы заблокируете его цикл обработки событий. С точки зрения пользователя, может показаться, что приложение зависает. Пользователь не сможет взаимодействовать с вашим приложением, пока оно находится в спящем режиме, используя этот метод. (В Windows вы даже можете получить предупреждение о том, что ваше приложение теперь не отвечает.)

К счастью, есть и другие методы, которые вы можете использовать, помимо time.sleep(). В следующих нескольких разделах вы узнаете, как добавлять вызовы Python sleep() как в Tkinter, так и в wxPython.

sleep в Tkinter

tkinter является частью стандартной библиотеки Python. Она может быть недоступна для вас, если вы используете предустановленную версию Python на Linux или Mac. Если вы получите ImportError, то вам нужно будет узнать, как добавить его в свою систему. Но если вы устанавливаете Python самостоятельно, то tkinter уже должен быть доступен.

Вы начнете с рассмотрения примера, в котором используется time.sleep(). Запустите этот код, чтобы увидеть, что произойдет, если вы добавите неправильный вызов Python sleep():

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()

После запуска кода нажмите кнопку в графическом интерфейсе пользователя. Кнопка будет удерживаться нажатой в течение трех секунд, ожидая завершения работы sleep(). Если бы в приложении были другие кнопки, вы бы не смогли их нажать. Вы также не можете закрыть приложение, пока оно находится в спящем режиме, поскольку оно не может отреагировать на событие закрытия.

Чтобы заставить tkinter нормально спать, вам необходимо использовать after():

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('I was delayed')

if __name__ == "__main__":
    root = tkinter.Tk()
    app = MyApp(root)
    root.mainloop()

Здесь вы создаете приложение размером 400 пикселей в ширину и 400 пикселей в высоту. В нем нет виджетов. Все, что он будет делать, это отображать рамку. Затем вы вызываете self.root.after(), где self.root является ссылкой на объект Tk(). after() принимает два аргумента:

1. Количество миллисекунд до перехода в спящий режим
2. Метод, который будет вызываться по завершении перехода в спящий режим

В этом случае ваше приложение выведет строку в стандартный вывод через 3 секунды. Вы можете рассматривать after() как tkinter версию time.sleep(), но она также добавляет возможность вызова функции после завершения режима ожидания.

Вы могли бы использовать эту функциональность для улучшения взаимодействия с пользователем. Добавив вызов Python sleep(), вы можете ускорить загрузку приложения, а затем запустить какой-либо более длительный процесс после его запуска. Таким образом, пользователю не придется ждать открытия приложения.

Спящий режим в wxPython

Между wxPython и Tkinter есть два основных различия:

1. В wxPython есть гораздо больше виджетов.
2. wxPython стремится выглядеть и чувствовать себя родным на всех платформах.

Фреймворк wxPython не входит в комплект поставки Python, поэтому вам нужно будет установить его самостоятельно. Если вы не знакомы с wxPython, ознакомьтесь с Как создать приложение с графическим интерфейсом Python с помощью wxPython.

В wxPython вы можете использовать wx.CallLater() для добавления вызова Python sleep():

import wx

class MyFrame(wx.Frame):
    def __init__(self):
        super().__init__(parent=None, title='Hello World')
        wx.CallLater(4000, self.delayed)
        self.Show()

    def delayed(self):
        print('I was delayed')

if __name__ == '__main__':
    app = wx.App()
    frame = MyFrame()
    app.MainLoop()

Здесь вы создаете подкласс wx.Frame напрямую, а затем вызываете wx.CallLater(). Эта функция принимает те же параметры, что и функция Tkinter after():

1. Количество миллисекунд до перехода в спящий режим
2. Метод, который будет вызываться по завершении перехода в спящий режим

Когда вы запустите этот код, вы должны увидеть небольшое пустое окно без каких-либо виджетов. Через 4 секунды вы увидите строку 'I was delayed', выведенную в стандартный вывод.

Одним из преимуществ использования wx.CallLater() является его потокобезопасность. Вы можете использовать этот метод из потока для вызова функции, которая находится в основном приложении wxPython.

Заключение

Благодаря этому руководству вы получили новую ценную технику, которую можно добавить в свой набор инструментов Python! Вы знаете, как добавлять задержки для ускорения работы ваших приложений и предотвращать использование ими системных ресурсов. Вы даже можете использовать вызовы Python sleep(), чтобы повысить эффективность перерисовки вашего графического интерфейса. Это значительно улучшит работу ваших клиентов!

Итак, вы узнали, как добавлять вызовы Python sleep() с помощью следующих инструментов:

• time.sleep()
• Decorators
• Threads
• asyncio
• Tkinter
• wxPython

Теперь вы можете воспользоваться тем, что узнали, и начать переводить свой код в режим ожидания!

Back to Top