Почему asyncio.current_task() не передается в функцию?

Question

[Dbtzhv]

async def info_start_tasks(message: types.Message, next_stage=None, delay=None, task=None):
        ...
        if next_stage!=None:
                task.cancel()

        ...
        end_task = current_task()
        print('CURRENT TASK', end_task)
        await info_start_tasks(message, task=end_task)

Ошибка:
'NoneType' object has no attribute 'cancel'

У функции (там рекурсия) есть сообщения, которые отправляются с asyncio.sleep, поэтому я хочу сделать так, что если функция вызывается с next_stage!=None, то эти отложенные сообщения не скидывались бы

Comments

[Everything_is_bad]

1. не нужно на ровном месте прибегать к рекурсии
2. код очень странный, лучше описывай проблему которую пытаешься решить

[Dbtzhv]

Everything_is_bad,

@dp.message_handler(filters.Text(startswith=['Продолжить выполнение', 'Попробовать еще раз']))
async def info_start_tasks(message: types.Message, next_stage=None, delay=None, task=None):
    try:
        user = get_object_or_404(User, telegram_id=message.chat.id)
        if user.life_counter < 1:
            raise TimeoutError
        funnel = Funnel.objects.get(members=user)
        if user.funnel_step == funnel.stage_id.count():
            raise ValueError
        if next_stage!=None:
            # task.cancel()
            print('NEXT STAGE:', next_stage)
            stages = list(funnel.stage_id.all())
            stage = Stage.objects.filter(stage_id=next_stage).first()
            index_of_next_stage = stages.index(stage)
            print('INDEX', index_of_next_stage)
            user.funnel_step = index_of_next_stage
            user.save()
        else:
            print('CHEEEECK344', user.funnel_step, funnel.stage_id.all())
            stage = list(funnel.stage_id.all())[user.funnel_step]
        statistics, _ = StageUser.objects.get_or_create(user=user, stage=stage)
        if stage.time_to_finish != 0:
            # logger.debug(f'запускаю таймер на {stage.time_to_finish * 3600} сек')
            if user.cache_delete:
                original_datetime = datetime.strptime(str(datetime.now()), "%Y-%m-%d %H:%M:%S.%f")
                user.start_time_point = original_datetime.strftime("%Y-%m-%dT%H:%M:%S")
                user.cache_delete = False
                user.save()
                current_datetime = datetime.now(datetime.UTC)
                statistics.time_to_starting_burn = datetime.now()
                statistics.save()
                data_to_1_burn = current_datetime + timedelta(hours=stage.time_to_finish * 0.5)
                data_to_2_burn = data_to_1_burn + timedelta(hours=stage.time_to_finish * 0.25)
                data_to_3_burn = data_to_2_burn + timedelta(hours=stage.time_to_finish * 0.25)
                task = check_stage_timeout_task.apply_async(
                    args=[statistics.id, [data_to_1_burn, data_to_2_burn, data_to_3_burn]],
                    eta=data_to_1_burn
                )
                TaskUser.objects.create(user=user, task_id=task.id)
        print('STAGE1:', stage)
        print(stage.message_id.all(), user.number_of_solved_tasks)
        answer_for_user_all = list(stage.message_id.all())#[user.number_of_solved_tasks]
        print('ANSWERRR', answer_for_user_all)
        for answer_for_user in answer_for_user_all:
            if delay==None:
                delay = answer_for_user.delay if answer_for_user.delay!=None else 0
            print('DELAY', delay)
            await asyncio.sleep(delay)
            delay=answer_for_user.delay if answer_for_user.delay!=None else 0
            print('STAGE2:', stage)
            await send_task_information(message.chat.id, answer_for_user, stage)
            await send_all_files(user, answer_for_user)
            await image_generation(answer_for_user, user)
        user.funnel_step+=1
        user.save()
        end_task = current_task()
        print('CURRENT TASK', end_task)
        await info_start_tasks(message, task=end_task)
        if (
            stage.is_generation
            and stage.description
            and stage.photo
            and user.number_of_solved_tasks == 0
        ):
            create_image(user, stage=stage)
            with open('send.png', "rb") as image_from_buffer:
                await bot.send_photo(
                    chat_id=user.telegram_id,
                    photo=image_from_buffer.read()
                    )
        if stage.linked:
            ilkeyboard = types.InlineKeyboardMarkup(row_width=4).add(types.InlineKeyboardButton(
                text='Ссылка для прохождения',
                url=f'{MAIN_URL}{funnel.owner}/{user}'
            ))
            await bot.send_message(
                chat_id=user.telegram_id,
                text='Этот этап необходимо пройти на сайте',
                reply_markup=ilkeyboard,
            )
        count = 0
        for button in answer_for_user.buttons.all():
            count += 1
            if button.button_type == 'Ссылка на ресурс':
                count -= 1
        if count == 0 and (not stage.with_open_answer or not (answer_for_user == list(stage.message_id.all())[-1])):
            user.number_of_solved_tasks += 1
            user.save()
            print('FUNC 44444')
            #await info_start_tasks(message, next_stage=next_stage)
    except Http404 as error:
        await handle_http4044_error(message.chat.id, error)
    except IndexError:
        await handle_index_error(user, funnel, message, next_stage)
    except BotBlocked:
        await bot_blocked_error(user, funnel)
    except TimeoutError:
        await handle_timeout_error(message.chat.id)
    except ValueError:
        answer, keyboard, _, _ = create_answer('Stage_matching_query_does_not_exist')
        return await bot.send_message(
            chat_id=user.telegram_id,
            text=answer.message_text.replace('<br>', '\n'),
            reply_markup=keyboard,
            parse_mode='HTML'
        )
    except Exception as error:
        await handle_exception(message.chat.id, error)

P.S. Это не моё творение, если что)

[Everything_is_bad]

Dbtzhv, тут всё очень плохо, кроме выше озвученного еще и синхротронные вызовы ORM, хотя в django ORM позволяет делать асинхронные вызовы.

[Dbtzhv]

Everything_is_bad, а что можно сделать?) Чтоб при вызове функции с аргументом next_stage, работа предыдущей с её asyncio.sleep() прерывалась

[Everything_is_bad]

Dbtzhv, убрать рекурсию, вызов части которую надо прервать, вынести из info_start_tasks и запустить отдельной задачей, которую уже и прерывать

Answer 1

[Vindicar]

Цитата из документации:

Return the currently running Task instance, or None if no task is running.

Как я это понимаю, если ты запланировал выполнение функции как задачи через asyncio.create_task(), то эта функция, и все, вызываемые в ней, смогут получить объект задачи, вызвав asyncio.current_task().
Если же управление в текущую функции было передано без использования задач на любом из уровней, только через await вызовы, то current_task() вернёт None.

Но вообще довольно странно. Эксперимент показывает, что хотя бы одна задача должна быть.

Вот код

import asyncio

async def print_task():
    print(asyncio.current_task())


async def foobar():
    await print_task()


async def main():
    print('Direct call')
    await foobar()
    print('create_task')
    task = asyncio.create_task(foobar())
    await task

# asyncio.run(main())  # даст такой же результат
asyncio.get_event_loop().run_until_complete(main())

На питоне 3.11 код выводит два объекта задачи, один создаётся run() или run_until_complete(), а другой - create_task(). Тогда получается, что получить None невозможно.