Выполнить асинхронную функцию синхронно

Question

Входные данные: Я пишу скрипт на python. Метод экземпляра класса из этого скрипта экспортируются в другой файл и помещается внутрь event loop, но этот event loop создаётся не мной, а фреймворком, поэтому я не могу в него впихнуть свою функцию. Код в примере максимально упрощён, а все функции заменены на выдуманные для того, чтобы было легче понять

Сам скрипт:

""" Импорт функции из одной библиотеки. Эта функция асинхронна и требует await перед собой
Мне необходимо выполнить эту функцию в своём скрипте """
from library import some_function

""" Импорт класса из фреймворка (никак не связано с библиотекой,
отткуда импортируется функция выше).
При запуске всей системы экземпляр этого класса импортируется в другой файл,
который я так же не могу изменять. В этом файле происходит создание и запуск event loop """
from framework import SomeClass

""" Создание экземпляра класса """
someclass = SomeClass()

""" Далее идёт изменение свойств класса и прочий код """
# ...
# ...
# ...

""" Здесь мне нужно выполнить функцию, которая была импортирована из библиотеки ранее НО: """
some_function() # - Ошибка! Функция не имеет await перед собой

await some_function() # - Ошибка! Нельзя использовать await вне функции

import asyncio
asyncio.run(some_function()) # - Ошибка! Вы пытаетесь создать event loop внутри другого event loop

Как быть?

Answer 1

Видимо, уже нашли решение. Вот для тех, кто ищет:

Используйте asyncio.get_event_loop() чтобы получить уже существующий event loop:

loop = asyncio.get_event_loop()
loop.run_until_complete(some_function()) # Будет ждать, пока some_function не закончит выполнение.

Согласно документации, run_until_complete выполняет async функцию, при этом блокируя исполнение программы в месте вызова. Другими словами, выполняет синхронную функцию синхронно.

Answer 2

Проблема была решена таких образом:

loop = asyncio.get_event_loop()
loop.create_task(some_function())

БЛОГ НА HUSL

Выполнить асинхронную функцию синхронно

Ответы (2 шт):