В прошлой статье мы рассматривали основные способы реализации асинхронного приложения в Python. Теперь же попробуем глубже разобраться в этом вопросе.
HTTP-запросы
Взаимодействие программы с сервером с помощью HTTP происходит довольно часто. И здесь также хорошо было бы использовать несколько потоков, чтобы не приходилось ждать, пока будет выполнена операция. В некоторых случаях приходится ждать немало, если есть перебои с сетью. Но даже если длительность запроса не сильно большая, пользователь все равно может заметить задержку.
Конечно, чтобы решить некоторые проблемы, можно воспользоваться готовыми решениями и пользоваться ими на этапе прототипирования. Тем не менее, это делать не рекомендуется, потому что они время от времени изменяются, и это может привести к сбою программы в самый неожиданный момент (например, если ее купит миллион пользователей).
Приведем пример, как реализация HTTP-коммуникации происходит с помощью стандартных методов Python.
import time
import json
import random
from http.server import HTTPServer, BaseHTTPRequestHandler
class RandomRequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
# Имитация задержки
time.sleep(3)
# Добавляем заголовки ответа
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
# Добавляем тело ответа
body = json.dumps({'random': random.random()})
self.wfile.write(bytes(body, "utf8"))
def main():
"""Запускает HTTP-сервер на порту 8090"""
server_address = ('', 8090)
httpd = HTTPServer(server_address, RandomRequestHandler)
httpd.serve_forever()
if __name__ == "__main__":
main()
В этом коде мы реализуем HTTP-сервер, выполняющий генерацию случайного ответа в формате JSON, после чего выводит его в графическом интерфейсе.
Чтобы запустить сервер, необходимо выполнить скрипт server.py и не останавливать этот процесс. Для получения запросов используется локальный порт 8090.
Клиентская часть программы включает метку для демонстрации информации пользователям и кнопки для выполнения нового запроса.
import json
import threading
import urllib.request
import tkinter as tk
class App(tk.Tk):
def __init__(self):
super().__init__()
self.title("Выполнение HTTP-запросов")
self.label = tk.Label(self, text="Нажмите 'Старт', чтобы получить случайное значение.")
self.button = tk.Button(self, text="Старт",
command=self.start_action)
self.label.pack(padx=60, pady=10)
self.button.pack(pady=10)
def start_action(self):
self.button.config(state=tk.DISABLED)
thread = AsyncAction()
thread.start()
self.check_thread(thread)
def check_thread(self, thread):
if thread.is_alive():
self.after(100, lambda: self.check_thread(thread))
else:
text = "Случайное значение: {}".format(thread.result)
self.label.config(text=text)
self.button.config(state=tk.NORMAL)
class AsyncAction(threading.Thread):
def run(self):
self.result = None
url = "http://localhost:8090"
with urllib.request.urlopen(url) as f:
obj = json.loads(f.read().decode("utf-8"))
self.result = obj["random"]
if __name__ == "__main__":
app = App()
app.mainloop()
Программа будет выглядеть таким образом. После того, как запрос будет закончен, клиентская часть покажет число, которое автоматическим образом генерируется на сервере. 
Если асинхронная операция выполняется, то кнопка, естественно, будет неактивной. Это помешает сделать новый запрос до того, как будет выполнен предыдущий. Хотя в некоторых ситуациях такая необходимость есть.
Работа HTTP-запросов
В данном примере мы расширили класс Thread, чтобы реализовать логику работы на сепарированном потоке с использованием любимого всеми объектно-ориентированного подхода. Чтобы это сделать, необходимо переопределить метод run(), который производит выполнение HTTP-запроса на локальный сервер.
class AsyncAction(threading.Thread): def run(self): # ...
Есть также большое количество клиентских библиотек, но в описанном примере применяется модуль urllib.request, который относится к стандартной библиотеке. В него включена функция urlopen(), аргументом которой служит адрес сайта (URL) в строчном виде.
Возвращаемое этой функцией значение – HTTP-ответ, который может использоваться в качестве контекстного менеджера. Таким образом, пользователь может безопасно получить информацию, а потом просто закрыть его с использованием ключевого слова with.
Возвращаемое значение сервером происходит в формате JSON-документа:
{«random»: 0.0915826359180778}
Чтобы его получить, необходимо в адресной строке браузера прописать http://localhost:8080 и нажать на клавишу Enter
Чтобы декорировать строку в объект, необходимо передать содержимое ответа функции loads(), которая относится к модулю json. Это позволит получить доступ к случайному значению и записать его в атрибуте result. Это позволит в случае ошибки избежать считывания этого атрибута.
def run(self):
self.result = None
url = "http://localhost:8090"
with urllib.request.urlopen(url) as f:
obj = json.loads(f.read().decode("utf-8"))
self.result = obj["random"]
Затем программа с графическим интерфейсом время от времени опрашивает статус потока, как было описано в прошлом фрагменте.
def check_thread(self, thread):
if thread.is_alive():
self.after(100, lambda: self.check_thread(thread))
else:
text = "Random value: {}".format(thread.result)
self.label.config(text=text)
self.button.config(state=tk.NORMAL)
Разница в том, что при неактивности потока, возможно получение значения result, так как еще до окончания выполнения оно было указано.
Использование прогрессбара для соединения потоков
Со шкалами прогресса сталкивался каждый, поскольку это – эффективный инструмент продемонстрировать, насколько выполнен тот или иной процесс. Выглядят они следующим образом.
С помощью шкал прогресса можно визуально показать, сколько процентов выполнено. Выражается это в проценты заполненной площади. Видим, что в приведенном примере процесс выполнен приблизительно на 25%, поскольку заполнена четверть всего поля.
С помощью прогрессбара можно дать пользователю обратную связь по поводу того, на каком этапе выполнения процесса находится приложение.
Наша программа увеличивает площадь заполнения шкалы прогресса каждый раз после нажатия кнопки «Старт».
Чтобы имитировать выполнение фоновой задачи, генерация инкремента будет осуществляться в другом потоке. На каждом шаге будет осуществляться приостановка выполнения на 1 секунду.
Настройка коммуникации осуществляется с использованием синхронизированной очереди, позволяющей выполнять обмен информацией, оберегая поток.
import time
import queue
import threading
import tkinter as tk
import tkinter.ttk as ttk
import tkinter.messagebox as mb
class App(tk.Tk):
def __init__(self):
super().__init__()
self.title("Пример Progressbar")
self.queue = queue.Queue()
self.progressbar = ttk.Progressbar(self, length=300,
orient=tk.HORIZONTAL)
self.button = tk.Button(self, text="Старт",
command=self.start_action)
self.progressbar.pack(padx=10, pady=10)
self.button.pack(padx=10, pady=10)
def start_action(self):
self.button.config(state=tk.DISABLED)
thread = AsyncAction(self.queue, 20)
thread.start()
self.poll_thread(thread)
def poll_thread(self, thread):
self.check_queue()
if thread.is_alive():
self.after(100, lambda: self.poll_thread(thread))
else:
self.button.config(state=tk.NORMAL)
mb.showinfo("Готово!", "Асинхронное действие завершено")
def check_queue(self):
while self.queue.qsize():
try:
step = self.queue.get(0)
self.progressbar.step(step * 100)
except queue.Empty:
pass
class AsyncAction(threading.Thread):
def __init__(self, queue, steps):
super().__init__()
self.queue = queue
self.steps = steps
def run(self):
for _ in range(self.steps):
time.sleep(1)
self.queue.put(1 / self.steps)
if __name__ == "__main__":
app = App()
app.mainloop()
Как отменять запланированные действия?
Можно не только планировать методы для откладывания, а и отменять те, которые находятся в очереди на выполнение. Типичный пример: если пользователю нужно подождать 10 секунд для того, чтобы программа проверила файлы на вирусы перед открытием. Если не хочется ждать, можно просто нажать клавишу «Стоп», и будут выполнять следующие предусмотренные куски кода.
Мы сделаем похожую программу, которая будет просто планировать действие через 5 секунд, а пользователь будет иметь возможность его остановить.
Для этого используется метод after_cancel(). Он использует идентификатор, который был возвращен при планировании действия с помощью метода after. В нашем случае это значение находится в атрибуте scheduled_id:
import tkinter as tk
class App(tk.Tk):
def __init__(self):
super().__init__()
self.button = tk.Button(self, command=self.start_action,
text="Подождите 5 секунд")
self.cancel = tk.Button(self, command=self.cancel_action,
text="Стоп", state=tk.DISABLED)
self.button.pack(padx=30, pady=20, side=tk.LEFT)
self.cancel.pack(padx=30, pady=20, side=tk.LEFT)
def start_action(self):
self.button.config(state=tk.DISABLED)
self.cancel.config(state=tk.NORMAL)
self.scheduled_id = self.after(5000, self.init_buttons)
def init_buttons(self):
self.button.config(state=tk.NORMAL)
self.cancel.config(state=tk.DISABLED)
def cancel_action(self):
print("Отмена событий", self.scheduled_id)
self.after_cancel(self.scheduled_id)
self.init_buttons()
if __name__ == "__main__":
app = App()
app.mainloop()
