Add more control on how updates acknowledged.

[Darkclainer]

aiogram version

3.x

Problem

Imagine you want to guarantee that you handle all messages with successful_payment. That is you
either somehow successfully responded to it (changed your state) or you want to receive and process
this update again (for example you crashed in the process).

Currently, you can implement this behavior for the long-polling if you use start_polling with
handle_as_tasks=False, or for the webhook if you use SimpleRequestHandler with
handle_in_background=False. The problem for me that in either case you control this behavior for
all updates.

For example with long-polling and disabled handle_as_tasks every single update will be handled
sequentially and after that the new request to Telegram will be made. I suggest to give user more
control how an individual update will be handled: sequentially or in the background. As an
alternative to solve the same problem, we can give user ability to process updates before they
handled: user can write updates to WAL and handle them on the next startup.

Possible solution

I will provide an example for the long-polling case.
The same idea can be applied for the webhook, but the implementation will be more intrusive.

Also, I'm not very experienced with writing backward compatible API in the Python.

Currently, start_polling has the next signature:

async def start_polling(
    self,
    *bots: Bot,
    polling_timeout: int = 10,
    handle_as_tasks: bool = True,
    backoff_config: BackoffConfig = DEFAULT_BACKOFF_CONFIG,
    allowed_updates: Optional[Union[List[str], UNSET_TYPE]] = UNSET,
    handle_signals: bool = True,
    close_bot_session: bool = True,
    **kwargs: Any,
) -> None:

I propose change to this:

UpdateHook = Callable[[Update], bool]

async def start_polling(
    self,
    *bots: Bot,
    polling_timeout: int = 10,
    handle_as_tasks: Union[bool, UpdateHook] = True,
    backoff_config: BackoffConfig = DEFAULT_BACKOFF_CONFIG,
    allowed_updates: Optional[Union[List[str], UNSET_TYPE]] = UNSET,
    handle_signals: bool = True,
    close_bot_session: bool = True,
    **kwargs: Any,
) -> None:

And change implementation of Dispatcher._polling from:

# ...
async for update in self._listen_updates(
    bot,
    polling_timeout=polling_timeout,
    backoff_config=backoff_config,
    allowed_updates=allowed_updates,
):
    handle_update = self._process_update(bot=bot, update=update, **kwargs)
    if handle_as_tasks:
        handle_update_task = asyncio.create_task(handle_update)
        self._handle_update_tasks.add(handle_update_task)
        handle_update_task.add_done_callback(self._handle_update_tasks.discard)
    else:
        await handle_update
# ...

To this (conceptually):

# ...
async for update in self._listen_updates(
    bot,
    polling_timeout=polling_timeout,
    backoff_config=backoff_config,
    allowed_updates=allowed_updates,
):
    handle_update = self._process_update(bot=bot, update=update, **kwargs)
    as_tasks = (
        handle_as_tasks
        if isinstance(handle_as_tasks, bool)
        else handle_as_tasks(update)
    )
    if as_tasks:
        handle_update_task = asyncio.create_task(handle_update)
        self._handle_update_tasks.add(handle_update_task)
        handle_update_task.add_done_callback(self._handle_update_tasks.discard)
    else:
        await handle_update
# ...

Alternatives

No response

Code example

Proposed change will allow handling specific messages sequentially or in the background:

dp = Dispatcher()
bot = Bot(token='example')
await dp.start_polling(bot, handle_as_tasks=lambda update: update.event_type != SuccessfulPayment)

or with WAL proposal:

wal = WAL()

def wal_write(u: Update):
    if u.event_type == SuccessfulPayment:
        wal.write(u)
    return True

dp = Dispatcher()
bot = Bot(token="example")
await dp.start_polling(bot, handle_as_tasks=wal_write)

Additional information

No response