Правильное обнаружение синтаксических ошибок с помощью сопрограмм в Python

#python #error-handling #async-await #python-asyncio

#python #обработка ошибок #асинхронное ожидание #python-asyncio

Вопрос:

Я пытаюсь написать небольшой скрипт на Python, который в основном выполняет две вещи параллельно. Я перехожу с JavaScript и пытаюсь использовать функции сопрограммы async / await в Python, но у меня возникли небольшие проблемы с тем, чтобы заставить его работать так, как я ожидаю.

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

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

Я попытался обернуть всю нарушающую сопрограмму в try / except , но это, похоже, нарушило другие мои сопрограммы (по-видимому, нарушив мои предположения о том, как работают сопрограммы).

 #!/usr/bin/env python3

import asyncio

async def other():
    while True:
        await asyncio.sleep(1)
        print('foobar')
        
async def test():
    try:
        while True:
            a = undefinedVariable
    except:
        print("Why doesn't this always catch??")

# catches as expected
asyncio.get_event_loop().run_until_complete(asyncio.wait([test()]))

# doesn't catch, fills screen with "foobar" lines, once a second, runs forever
asyncio.get_event_loop().run_until_complete(asyncio.wait([test(), other()]))

# Trying other things...
# Error:
# RuntimeError: no running event loop
# sys:1: RuntimeWarning: coroutine 'test' was never awaited"
asyncio.get_event_loop().run_until_complete(asyncio.wait([asyncio.create_task(t()) for t in [test, other]]))
  

Запуск одной проблемной сопрограммы работает так, как ожидалось.

Объединение его с другой сопрограммой, которая не выдает ошибок, приводит к тому, что другая сопрограмма перестает уважать asyncio.sleep .

Однажды я дошел до того, что, если я нажму Ctrl C, он перестанет печатать «foobar», напечатает синтаксическую ошибку и затем завершит работу, но мне не удалось захватить этот код, и он потерян для отмены.

Я чувствую, что мне не хватает чего-то простого. Любые указатели будут с благодарностью приняты.

Комментарии:

1. Упс! Что ж, это объясняет неуправляемый цикл в моем примере. Не объясняет остальную часть моей проблемы (что происходит с await ). Образец кода обновлен.

2. Я пытаюсь выполнить "Why doesn't this always catch??" печать один раз, а затем выполнить остальную часть выполнения. В нынешнем виде я не понял, как заставить его остановить other() задачу. Похоже, что еще одно нажатие дало мне ответ, который работает, по крайней мере, для меня.

Ответ №1:

Я нашел ссылку в документах, которая asyncio.run() должна использоваться в качестве «основной» точки входа и обрабатывает все создание и очистку цикла, поэтому я попробовал.

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

 asyncio.run(asyncio.wait([test(), other()], return_when=asyncio.FIRST_COMPLETED))
  

Чего я не понимаю, так это почему, когда я пробовал return_when=asyncio.FIRST_COMPLETED раньше, он продолжал работать. Возможно, была какая-то другая опечатка, которую я пропустил.

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

Не уверен, поможет ли это кому-то еще в будущем. Спасибо всем, кто посмотрел.