Polling updates#

API internals#

glQiwiApi.core.event_fetching.executor.start_polling(wallet: Union[glQiwiApi.qiwi.clients.wallet.client.QiwiWallet, glQiwiApi.qiwi.clients.wrapper.QiwiWrapper], dispatcher: glQiwiApi.core.event_fetching.dispatcher.BaseDispatcher, *plugins: glQiwiApi.plugins.abc.Pluggable, skip_updates: bool = False, timeout_in_seconds: float = 5, on_startup: Optional[Callable[[Context], Union[None, Awaitable[None]]]] = None, on_shutdown: Optional[Callable[[Context], Union[None, Awaitable[None]]]] = None, loop: Optional[asyncio.events.AbstractEventLoop] = None, context: Optional[Union[Dict[str, Any], glQiwiApi.core.event_fetching.executor.Context]] = None) → None[source]#

Setup for long-polling mode. Support only glQiwiApi.types.Transaction as event.

Parameters

  • wallet – instance of QiwiWrapper

  • dispatcher –

  • skip_updates –

  • timeout_in_seconds – timeout of polling in seconds, if the timeout is too small, the API can throw an exception

  • on_startup – function or coroutine, which will be executed on startup

  • on_shutdown – function or coroutine, which will be executed on shutdown

  • plugins – List of plugins, that will be executed together with polling. For example builtin TelegramPollingPlugin or other class, that implement Pluggable abc interface, deal with foreign framework/application in the background

  • loop –

  • context – context, that could be transmitted to handlers

async glQiwiApi.core.event_fetching.executor.start_non_blocking_qiwi_api_polling(wallet: Union[glQiwiApi.qiwi.clients.wallet.client.QiwiWallet, glQiwiApi.qiwi.clients.wrapper.QiwiWrapper], dispatcher: glQiwiApi.core.event_fetching.dispatcher.BaseDispatcher, skip_updates: bool = False, timeout_in_seconds: float = 5, on_startup: Optional[Callable[[Context], Union[None, Awaitable[None]]]] = None, on_shutdown: Optional[Callable[[Context], Union[None, Awaitable[None]]]] = None, loop: Optional[asyncio.events.AbstractEventLoop] = None, context: Optional[Union[Dict[str, Any], glQiwiApi.core.event_fetching.executor.Context]] = None) → _asyncio.Task[source]#
glQiwiApi.core.event_fetching.executor.configure_app_for_qiwi_webhooks(wallet: Union[glQiwiApi.qiwi.clients.wallet.client.QiwiWallet, glQiwiApi.qiwi.clients.wrapper.QiwiWrapper], dispatcher: glQiwiApi.core.event_fetching.dispatcher.BaseDispatcher, app: aiohttp.web_app.Application, cfg: glQiwiApi.core.event_fetching.webhooks.config.WebhookConfig) → aiohttp.web_app.Application[source]#
class glQiwiApi.core.event_fetching.executor.BaseExecutor(dispatcher: glQiwiApi.core.event_fetching.dispatcher.BaseDispatcher, *plugins: glQiwiApi.plugins.abc.Pluggable, context: glQiwiApi.core.event_fetching.executor.Context, loop: Optional[asyncio.events.AbstractEventLoop] = None, on_startup: Optional[Callable[[...], Any]] = None, on_shutdown: Optional[Callable[[...], Any]] = None)[source]#

Bases: abc.ABC

__init__(dispatcher: glQiwiApi.core.event_fetching.dispatcher.BaseDispatcher, *plugins: glQiwiApi.plugins.abc.Pluggable, context: glQiwiApi.core.event_fetching.executor.Context, loop: Optional[asyncio.events.AbstractEventLoop] = None, on_startup: Optional[Callable[[...], Any]] = None, on_shutdown: Optional[Callable[[...], Any]] = None) → None[source]#
property loop: asyncio.events.AbstractEventLoop#
async welcome() → None[source]#
async goodbye() → None[source]#
class glQiwiApi.core.event_fetching.executor.PollingExecutor(wallet: Union[glQiwiApi.qiwi.clients.wallet.client.QiwiWallet, glQiwiApi.qiwi.clients.wrapper.QiwiWrapper], dispatcher: glQiwiApi.core.event_fetching.dispatcher.BaseDispatcher, *plugins: glQiwiApi.plugins.abc.Pluggable, context: glQiwiApi.core.event_fetching.executor.Context, loop: Optional[asyncio.events.AbstractEventLoop] = None, timeout: Union[float, int] = 5, skip_updates: bool = False, on_startup: Optional[Callable[[Context], Union[None, Awaitable[None]]]] = None, on_shutdown: Optional[Callable[[Context], Union[None, Awaitable[None]]]] = None)[source]#

Bases: glQiwiApi.core.event_fetching.executor.BaseExecutor

__init__(wallet: Union[glQiwiApi.qiwi.clients.wallet.client.QiwiWallet, glQiwiApi.qiwi.clients.wrapper.QiwiWrapper], dispatcher: glQiwiApi.core.event_fetching.dispatcher.BaseDispatcher, *plugins: glQiwiApi.plugins.abc.Pluggable, context: glQiwiApi.core.event_fetching.executor.Context, loop: Optional[asyncio.events.AbstractEventLoop] = None, timeout: Union[float, int] = 5, skip_updates: bool = False, on_startup: Optional[Callable[[Context], Union[None, Awaitable[None]]]] = None, on_shutdown: Optional[Callable[[Context], Union[None, Awaitable[None]]]] = None) → None[source]#
start_polling() → None[source]#
async start_non_blocking_polling() → _asyncio.Task[source]#
async process_updates(history: glQiwiApi.qiwi.clients.wallet.types.transaction.History) → None[source]#

Guidelines#

This section explains how to properly poll transactions from QIWI API.

You can’t help registering handlers and start polling, so in example above it is shown how to do it rightly. Lets do it with decorators:

️So, we also have to import executor and pass on our client, that contains functions start_polling and start_webhook.

from glQiwiApi import QiwiWallet
from glQiwiApi.core.event_fetching import executor
from glQiwiApi.core.event_fetching.dispatcher import QiwiDispatcher
from glQiwiApi.core.event_fetching.executor import Context
from glQiwiApi.core.event_fetching.filters import ExceptionFilter
from glQiwiApi.qiwi.clients.wallet.types import Transaction
from glQiwiApi.qiwi.exceptions import QiwiAPIError

qiwi_dp = QiwiDispatcher()
wallet = QiwiWallet(api_access_token='token', phone_number='+phone number')


@qiwi_dp.transaction_handler()
async def handle_transaction(t: Transaction, ctx: Context):
    """Handle transaction here"""
    ctx.wallet  # this way you can use QiwiWallet instance to avoid global variables


@qiwi_dp.exception_handler(ExceptionFilter(QiwiAPIError))
async def handle_exception(err: QiwiAPIError, ctx: Context):
    pass


if __name__ == '__main__':
    executor.start_polling(wallet, qiwi_dp)

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. 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.

from glQiwiApi import QiwiWallet
from glQiwiApi.core.event_fetching import executor
from glQiwiApi.core.event_fetching.dispatcher import QiwiDispatcher
from glQiwiApi.core.event_fetching.executor import Context
from glQiwiApi.qiwi.clients.wallet.types import Transaction

qiwi_dp = QiwiDispatcher()
wallet = QiwiWallet(api_access_token='token', phone_number='+phone number')


@qiwi_dp.transaction_handler()
async def handle_transaction(t: Transaction, ctx: Context):
    """Handle transaction here"""
    ctx.wallet  # this way you can use QiwiWallet instance to avoid global variables


async def on_startup(ctx: Context):
    ctx.wallet  # do something here


async def on_shutdown(ctx: Context):
    pass


if __name__ == '__main__':
    executor.start_polling(wallet, qiwi_dp, on_startup=on_startup, on_shutdown=on_shutdown)

Make aiogram work with glQiwiApi#

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.

from aiogram import Bot, Dispatcher
from aiogram.types import Message

from glQiwiApi import QiwiWallet
from glQiwiApi.core.event_fetching import executor
from glQiwiApi.core.event_fetching.dispatcher import QiwiDispatcher
from glQiwiApi.core.event_fetching.executor import Context
from glQiwiApi.plugins import AiogramPollingPlugin
from glQiwiApi.qiwi.clients.wallet.types import Transaction

qiwi_dp = QiwiDispatcher()
wallet = QiwiWallet(api_access_token='token', phone_number='+phone number')

dp = Dispatcher(Bot('BOT TOKEN'))


@qiwi_dp.transaction_handler()
async def handle_transaction(t: Transaction, ctx: Context):
    """Handle transaction here"""
    ctx.wallet  # this way you can use QiwiWallet instance to avoid global variables


@dp.message_handler()
async def handle_message(msg: Message):
    await msg.answer(text='Hello world')


if __name__ == '__main__':
    executor.start_polling(wallet, qiwi_dp, AiogramPollingPlugin(dp))

Alternatively you can run polling at on_startup event

from aiogram import Bot, Dispatcher
from aiogram.types import Message
from aiogram.utils import executor

from glQiwiApi import QiwiWallet
from glQiwiApi.core.event_fetching.dispatcher import QiwiDispatcher
from glQiwiApi.core.event_fetching.executor import Context, start_non_blocking_qiwi_api_polling
from glQiwiApi.qiwi.clients.wallet.types import Transaction

qiwi_dp = QiwiDispatcher()
wallet = QiwiWallet(api_access_token='token', phone_number='+phone number')

dp = Dispatcher(Bot('BOT TOKEN'))


@qiwi_dp.transaction_handler()
async def handle_transaction(t: Transaction, ctx: Context):
    """Handle transaction here"""


@dp.message_handler()
async def handle_message(msg: Message):
    await msg.answer(text='Hello world')


async def on_startup(dp: Dispatcher):
    await start_non_blocking_qiwi_api_polling(wallet, qiwi_dp)


if __name__ == '__main__':
    executor.start_polling(dp, on_startup=on_startup)

Example usage without global variables#

import logging
from typing import cast

from aiogram import Bot, Dispatcher, types

from glQiwiApi import QiwiWrapper
from glQiwiApi.core.event_fetching import executor
from glQiwiApi.core.event_fetching.dispatcher import QiwiDispatcher
from glQiwiApi.core.event_fetching.executor import Context
from glQiwiApi.plugins import AiogramPollingPlugin
from glQiwiApi.qiwi.clients.wallet.types import Transaction

api_access_token = 'token'
phone_number = '+phone number'

logger = logging.getLogger(__name__)


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


async def qiwi_transaction_handler(update: Transaction, ctx: Context):
    print(update)


def on_startup(ctx: Context) -> None:
    logger.info('This message logged on startup')
    register_handlers(ctx)


def register_handlers(ctx: Context):
    ctx['qiwi_dp'].transaction_handler()(qiwi_transaction_handler)
    dispatcher = cast(Dispatcher, ctx['dp'])
    dispatcher.register_message_handler(aiogram_message_handler)


def run_application() -> None:
    logging.basicConfig(level=logging.INFO)
    bot = Bot('BOT TOKEN')
    dp = Dispatcher(bot)
    wallet = QiwiWrapper(api_access_token=api_access_token, phone_number=phone_number)
    qiwi_dp = QiwiDispatcher()

    executor.start_polling(
        wallet,
        qiwi_dp,
        AiogramPollingPlugin(dp),
        on_startup=on_startup,
        skip_updates=True,
        context={'dp': dp, 'qiwi_dp': qiwi_dp},
    )


if __name__ == '__main__':
    run_application()