Как использовать AsyncIOScheduler в Django?

Я разрабатываю API с использованием Django Rest Framework, который периодически вызывает асинхронный метод через AsyncIOScheduler (компонент APScheduler), заставляя его выполняться каждые несколько секунд. Однако это не работает так, как ожидалось, и в консоль ничего не выводится. Ниже приведен мой код.

Определение планировщика:

scheduler = AsyncIOScheduler()

log = logging.getLogger(__name__)

def start_scheduler():
    if not scheduler.running:
        scheduler.start()
        logging.info('scheduler started')

Код для запуска планировщика:

class ApiConfig(AppConfig):
    name = 'api'

    def ready(self):
        from .scheduler import start_scheduler
        start_scheduler()

Определение конечной точки интерфейса API:

class Boo():

    async def test():
        print("testFoo")
        await asyncio.sleep(1)
        print("testFoo done")
        
class Foo():
    def __init__(self):
        self.insideObjs = [Boo() for _ in range(10)]

    async def run(self):
        tasks = [boo.test() for boo in self.insideObjs]

        logging.info(f'waiting for jobs done')
        results = await asyncio.gather(*tasks)
        logging.info(f'all jobs done')

class FooViewSet(viewsets.ViewSet):

    @action(detail=False, methods=['post'], url_path='start')
    def start(self, request):
        foo = Foo()
        scheduler.add_job(foo.run, 'interval', seconds=3,
                          id="123", replace_existing=True)

По моим ожиданиям, 'testFoo' и 'testFoo done' должны выводиться в консоль каждые 3 секунды, но этого не происходит. Я попытался изменить функцию ready() класса ApiConfig на:

class ApiConfig(AppConfig):
    name = 'api'

    def ready(self):
        from .scheduler import start_scheduler
        start_scheduler()

до

class ApiConfig(AppConfig):
    name = 'api'

    def ready(self):
        from .scheduler import start_scheduler
        start_scheduler()

        try:
            asyncio.get_event_loop().run_forever()
        except (KeyboardInterrupt, SystemExit):
            pass

Однако после этого изменения сервер Django вечно находится в ожидании, не получая ни одного запроса.

Я также попытался добавить код цикла событий в manage.py:

def main():
    """Run administrative tasks."""
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'foo.settings')
    try:
        from django.core.management import execute_from_command_line
    except ImportError as exc:
        raise ImportError(
            "Couldn't import Django. Are you sure it's installed and "
            "available on your PYTHONPATH environment variable? Did you "
            "forget to activate a virtual environment?"
        ) from exc
    execute_from_command_line(sys.argv)

    try:
        asyncio.get_event_loop().run_forever()
    except (KeyboardInterrupt, SystemExit):
        pass

if __name__ == "__main__":
    main()

При таком изменении строка asyncio.get_event_loop().run_forever() никогда не будет выполняться.

Как заставить 10 объектов Boo в Foo запускаться одновременно каждые 3 секунды?

Вернуться на верх