Polling updates

Tip

👩🏻🎨 start_polling has the same signature as start_webhook “under the hood”

Guidelines

👩🏻🔬This API method gives a chance to hook updates without webhooks on your machine, but it’s possible to use both approaches anyway.

🧑🎓 First of all, before starting polling, you need to register handlers, just like when installing webhooks. Lets do it with decorators, but you also can do it another way, using wallet.register_transaction_handler, wallet.register_bill_handler or wallet.register_error_handler

Add handlers

import logging

from glQiwiApi import QiwiWrapper, types
from glQiwiApi.utils import executor

api_access_token = "your token"
phone_number = "your number"

wallet = QiwiWrapper(api_access_token=api_access_token, phone_number=phone_number)

logger = logging.getLogger(__name__)


@wallet.transaction_handler()
async def my_first_handler(update: types.Transaction):
    assert isinstance(update, types.Transaction)


def on_startup(wrapper: QiwiWrapper):
    logger.info("This message logged on startup")


executor.start_polling(wallet, on_startup=on_startup, skip_updates=True)

🧙♀️So, we also have to import executor Executor overview and pass on our client, that contains user-friendly functions start_polling and start_webhook.

import executor module

import logging

from glQiwiApi import QiwiWrapper, types
from glQiwiApi.utils import executor

api_access_token = "your token"
phone_number = "your number"

wallet = QiwiWrapper(api_access_token=api_access_token, phone_number=phone_number)

logger = logging.getLogger(__name__)


@wallet.transaction_handler()
async def my_first_handler(update: types.Transaction):
    assert isinstance(update, types.Transaction)


def on_startup(wrapper: QiwiWrapper):
    logger.info("This message logged on startup")


executor.start_polling(wallet, on_startup=on_startup, skip_updates=True)

Events

👨🔬 Then, you can start polling, but, let’s make it clear which arguments you should pass on to start_polling function. You can also specify events like on_shutdown or on_startup.

import logging

from glQiwiApi import QiwiWrapper, types
from glQiwiApi.utils import executor

api_access_token = "your token"
phone_number = "your number"

wallet = QiwiWrapper(api_access_token=api_access_token, phone_number=phone_number)

logger = logging.getLogger(__name__)


@wallet.transaction_handler()
async def my_first_handler(update: types.Transaction):
    assert isinstance(update, types.Transaction)


def on_startup(wrapper: QiwiWrapper):
    logger.info("This message logged on startup")


executor.start_polling(wallet, on_startup=on_startup, skip_updates=True)

😼 As you can see, in the example we have a function that we pass as an argument to on_startup. As you may have guessed, this function will be executed at the beginning of the polling.

import logging

from glQiwiApi import QiwiWrapper, types
from glQiwiApi.utils import executor

api_access_token = "your token"
phone_number = "your number"

wallet = QiwiWrapper(api_access_token=api_access_token, phone_number=phone_number)

logger = logging.getLogger(__name__)


@wallet.transaction_handler()
async def my_first_handler(update: types.Transaction):
    assert isinstance(update, types.Transaction)


def on_startup(wrapper: QiwiWrapper):
    logger.info("This message logged on startup")


executor.start_polling(wallet, on_startup=on_startup, skip_updates=True)

Aiogram + glQiwiApi = friends🤩

🧚♀️ Also, you can very easily implement simultaneous polling of updates from both aiogram and QIWI API.

In the example below, we catch all text messages and return the same “Hello” response.

polling together with aiogram

import logging

from aiogram import Bot, Dispatcher
from aiogram import types

from glQiwiApi import QiwiWrapper
from glQiwiApi.plugins.telegram.polling import TelegramPollingPlugin
from glQiwiApi.types import Transaction
from glQiwiApi.utils import executor

api_access_token = "your token"
phone_number = "your number"

bot = Bot("token from BotFather")
dp = Dispatcher(bot)

wallet = QiwiWrapper(api_access_token=api_access_token, phone_number=phone_number)

logger = logging.getLogger(__name__)


@dp.message_handler()
async def message_handler(msg: types.Message):
    await msg.answer(text="Привет😇")


@wallet.transaction_handler()
async def my_first_handler(update: Transaction):
    assert isinstance(update, Transaction)


def on_startup(wrapper: QiwiWrapper):
    logger.info("This message logged on startup")


executor.start_polling(wallet, TelegramPollingPlugin(dp), on_startup=on_startup)

Error handling

from __future__ import annotations

import logging
from typing import Any

from glQiwiApi import QiwiWrapper
from glQiwiApi.utils import executor

api_access_token = "your token"
phone_number = "your number"

wallet = QiwiWrapper(api_access_token=api_access_token, phone_number=phone_number)
logger = logging.getLogger(__name__)


@wallet.error_handler(exception=OSError)
async def os_error_handler(exception: Exception, *args: Any) -> None:
    # do something here, if exception has occurred, for example log it
    logger.exception("Exception %s occurred. WTF!!!", exception)


executor.start_polling(wallet, skip_updates=True)

Example usage without global variables😇

import logging
from typing import cast

from aiogram import Bot, Dispatcher
from aiogram import types

from glQiwiApi import QiwiWrapper
from glQiwiApi.plugins.telegram.polling import TelegramPollingPlugin
from glQiwiApi.types import Transaction
from glQiwiApi.utils import executor

api_access_token = "your token"
phone_number = "+your number"

logger = logging.getLogger(__name__)


def run_application() -> None:
    bot = Bot("token from BotFather")
    dp = Dispatcher(bot)
    wallet = QiwiWrapper(api_access_token=api_access_token, phone_number=phone_number)

    # set dispatcher to wallet instance for register aiogram handlers
    wallet["dispatcher"] = dp

    executor.start_polling(wallet, TelegramPollingPlugin(dp), on_startup=on_startup, skip_updates=True)


def register_handlers(wrapper: QiwiWrapper):
    wrapper.register_transaction_handler(qiwi_transaction_handler)
    dispatcher = cast(Dispatcher, wrapper["dispatcher"])
    dispatcher.register_message_handler(aiogram_message_handler)


async def aiogram_message_handler(msg: types.Message):
    await msg.answer(text="Привет😇")


async def qiwi_transaction_handler(update: Transaction):
    assert isinstance(update, Transaction)
    print(update)


def on_startup(wrapper: QiwiWrapper) -> None:
    logger.info("This message logged on startup")
    register_handlers(wrapper)


if __name__ == '__main__':
    run_application()