Группы асинхронных задач в Python 3.11

Вчера на официальном сайте был опубликован первый релиз-кандидат Python 3.11, который принесет важные оптимизации и доработки в возможности языка. Релиз планируется в октябре этого года, но уже сейчас можно поэкспериментировать с новыми возможностями и сегодня мы поговорим о группах исключений и асинхронных задач. Первые позволяют одновременно выбрасывать и обрабатывать несколько исключений, в то время как вторые позволяют объединять задачи в общий event loop и координированно управлять группами задач.

Для тестирования мы будем использовать образ контейнера python:3.11-rc-slim-buster. Запустим контейнер через docker run --rm -it python:3.11-rc-slim-buster и получим доступ к REPL, где мы можем экспериментировать с новыми языковыми возможностями. Также мы сможем запускать python-файл через cat test.py | docker run -i python:3.11-rc-slim-buster.

Начнем с групп исключений и сразу попробуем подготовить для них пример. Группы исключений позволяют обрабатывать несколько одновременно возникших исключений (например, в async-функциях) и интерпретировать их как список объектов (ранее обрабатывалось только первое исключение). Более подробно спецификация групп исключений описана здесь.

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

В Python 3.11 добавлен новый класс ExceptionGroup, который может сообщать о возникновении нескольких исключений и давать для них текстовое описание. Это добавляет контекста к сообщению об ошибке и позволяет объединять несколько событий (кроме того, группа исключений может содержать другие группы и, таким образом, исключения могут быть рекурсивными). Если исключение не будет обработано, мы получим трассировку ошибки, в которой дерево исключений будет показано с учетом вложенности и комментариев. При этом при перехвате группы через try-except будет обнаружен только внешний объект ExceptionGroup, но не связанные с ним исключения. Для корректной обработки одного из исключений группы используется новая синтаксическая конструкция try - except*. Например, если выбросить одновременно исключения ValueError и ZeroDivisionError, то обработка их может выполнять независимо (и параллельно):

try:
  raise ExceptionGroup('Multiple exceptions', [ValueError(), ZeroDivisionError()])
except* ValueError as gr:
	print('Value error '+gr.exceptions)
except* ZeroDivisionError as gr:
  print('Zero division error '+gr.exceptions)

Обратите внимание, что в объект ошибки собираются все экземпляры данного типа (например, может быть отправлено несколько ValueError и можно получить доступ к каждому из них из списка exceptions). Если не все отправленные типы были обработаны в except*, они будут отправлены выше (и могут привести к аварийному завершению, при этом в трассировке будут указаны только необработанные типы).

В Python 3.11 все типы единичных исключений будут автоматически распознаваться в except*, как если бы они были отправлены внутри группы исключений.

Наиболее интересный сценарий - возникновение последовательных исключений при ожидании завершения группы задач. Предположим, что у нас есть относительно долгая задача, которая внутри себя обрабатывает данные из файла, которая может вызвать исключение. Тогда что будет возвращено, если мы вызовем несколько задач и соберем их результаты через asyncio.gather?

import asyncio
import io

async def process_file(filename):
  with open(filename, mode="r"):
  	print('File is opened')
    return True

async def process(filenames):
	tasks = [asyncio.create_task(process_file(filename)) for filename in filenames]
  await asyncio.gather(*tasks)
  
if __name__=="__main__":
	asyncio.run(process(['file1','file2','file3'])

Запустим это приложение (в каталоге, где нет файлов file1, file2 и file3) и увидим, что в трассировке будет отмечена только одна ошибка. Чтобы обойти эту проблему мы можем использовать флаг return_exceptions в gather, но более хорошим решением будет использование asyncio.TaskGroup. При выполнении задач в группе, исключения объединяются в общую ExceptionGroup и могут быть обработаны с помощью except*. Управление ресурсами (например, создание задачи) теперь будет выполняться внутри группы (вместо asyncio.create_task будет использовать tg.create_task), которая создается через менеджер контекста. Например, рассмотренный выше код может быть переработан следующим образом:

import asyncio

async def process_file(filename):
  with open(filename, mode="r"):
    print('File is opened')
    return True

async def process(filenames):
  try:
    async with asyncio.TaskGroup() as tg:
      tasks = [tg.create_task(process_file(filename)) for filename in filenames]
  except* FileNotFoundError as errors:
    print(f'Files not found')
    print([e.filename for e in errors.exceptions])
    
if __name__=="__main__":
	asyncio.run(process(['file1','file2','file3'])

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

cat test.py | docker run -i python:3.11-rc-slim-buster
Files not found
['file1', 'file2', 'file3']

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

И в заключение приглашаю всех на бесплатный урок по теме: "Чистая архитектура в Python разработке", который проведет мой коллега из OTUS - Станислав Ступников.

• Зарегистрироваться на бесплатный урок