kamihi.tg.client

Telegram client module.

This module provides a Telegram client for sending messages and handling commands.

License

MIT

Examples:

>>> from kamihi.tg.client import TelegramClient
>>> from kamihi.base.config import KamihiSettings
>>> client = TelegramClient(KamihiSettings(), [])
>>> client.run()

Classes:

Name	Description
TelegramClient	Telegram client class.

TelegramClient

TelegramClient(
    settings: KamihiSettings, handlers: list[BaseHandler]
)

Telegram client class.

This class provides methods to send messages and handle commands.

Initialize the Telegram client.

Parameters:

Name	Type	Description	Default
settings	KamihiSettings	The settings object.	required
handlers	list[BaseHandler]	List of handlers to register.	required

Methods:

Name	Description
register_run_once_job	Add a job to run once.
reset_scopes	Reset the command scopes for the bot.
run	Run the Telegram bot.
set_scopes	Set the command scopes for the bot.
stop	Stop the Telegram bot.

Source code in src/kamihi/tg/client.py

def __init__(self, settings: KamihiSettings, handlers: list[BaseHandler]) -> None:
    """
    Initialize the Telegram client.

    Args:
        settings (KamihiSettings): The settings object.
        handlers (list[BaseHandler]): List of handlers to register.

    """
    self._bot_settings = settings

    if self._bot_settings.testing:
        self._base_url = "https://api.telegram.org/bot{token}/test"

    # Set up the application with all the settings
    self._builder = Application.builder()
    self._builder.base_url(self._base_url)
    self._builder.token(settings.token)
    self._builder.defaults(
        Defaults(
            tzinfo=settings.timezone_obj,
            parse_mode=ParseMode.MARKDOWN_V2,
        )
    )
    self._builder.post_init(_post_init)
    self._builder.post_shutdown(_post_shutdown)
    self._builder.persistence(DictPersistence(bot_data_json=settings.model_dump_json()))

    # Build the application
    self._app: Application = self._builder.build()

    # Register the handlers
    for handler in handlers:
        with logger.catch(exception=TelegramError, message="Failed to register handler"):
            self._app.add_handler(handler)

    # Register the default handlers
    if settings.responses.default_enabled:
        self._app.add_handler(MessageHandler(filters.TEXT, default), group=1000)
    self._app.add_error_handler(error)

register_run_once_job

register_run_once_job(
    callback: callable, when: int
) -> None

Add a job to run once.

Parameters:

Name	Type	Description	Default
callback	callable	The callback function to run.	required
when	int	second from now to run the job.	required

Source code in src/kamihi/tg/client.py

def register_run_once_job(self, callback: callable, when: int) -> None:
    """
    Add a job to run once.

    Args:
        callback (callable): The callback function to run.
        when (int): second from now to run the job.

    """
    self._app.job_queue.run_once(callback, when)

reset_scopes async

reset_scopes(context: CallbackContext) -> None

Reset the command scopes for the bot.

This method clears all command scopes and sets the default commands.

Parameters:

Name	Type	Description	Default
context	CallbackContext	The context of the callback. Not used but required for this function to be registered as a job.	required

Source code in src/kamihi/tg/client.py

async def reset_scopes(self, context: CallbackContext) -> None:  # noqa: ARG002
    """
    Reset the command scopes for the bot.

    This method clears all command scopes and sets the default commands.

    Args:
        context (CallbackContext): The context of the callback. Not used but required for
            this function to be registered as a job.

    """
    if self._bot_settings.testing:
        logger.debug("Testing mode, skipping resetting scopes")
        return

    with logger.catch(exception=TelegramError, message="Failed to reset scopes"):
        await self._app.bot.set_my_commands(commands=[])
        logger.debug("Scopes erased")

run

run() -> None

Run the Telegram bot.

Source code in src/kamihi/tg/client.py

def run(self) -> None:
    """Run the Telegram bot."""
    logger.trace("Starting main loop...")
    self._app.run_polling(allowed_updates=Update.ALL_TYPES)

set_scopes async

set_scopes(scopes: dict[int, list[BotCommand]]) -> None

Set the command scopes for the bot.

Parameters:

Name	Type	Description	Default
scopes	dict[int, list[BotCommand]]	The command scopes to set.	required

Source code in src/kamihi/tg/client.py

async def set_scopes(self, scopes: dict[int, list[BotCommand]]) -> None:
    """
    Set the command scopes for the bot.

    Args:
        scopes (dict[int, list[BotCommand]]): The command scopes to set.

    """
    if self._bot_settings.testing:
        logger.debug("Testing mode, skipping setting scopes")
        return

    for user_id, commands in scopes.items():
        lg = logger.bind(user_id=user_id, commands=[command.command for command in commands])
        with lg.catch(
            exception=TelegramError,
            message="Failed to set scopes for user {user_id}",
        ):
            await self._app.bot.set_my_commands(
                commands=commands,
                scope=BotCommandScopeChat(user_id),
            )
            lg.debug("Scopes set")

stop async

stop() -> None

Stop the Telegram bot.

Source code in src/kamihi/tg/client.py

async def stop(self) -> None:
    """Stop the Telegram bot."""
    logger.trace("Stopping main loop...")
    await self._app.stop()