kamihi.tg.client

Telegram client module.

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

License

MIT

Classes:

Name	Description
TelegramClient	Telegram client class.

TelegramClient

TelegramClient(
    _post_init: Callable, _post_shutdown: Callable
)

Telegram client class.

This class provides methods to send messages and handle commands.

Initialize the Telegram client.

Parameters:

Name	Type	Description	Default
_post_init	callable	Function to call after the application is initialized.	required
_post_shutdown	callable	Function to call after the application is shut down.	required

Methods:

Name	Description
add_datasources	Add data sources to the Telegram client.
add_default_handlers	Add default handlers to the Telegram client.
add_handlers	Add handlers to the Telegram client.
add_jobs	Add jobs to the Telegram client.
add_pages_handler	Add the pages handler to the Telegram client.
reset_scopes	Reset the command scopes for the bot.
run	Run the Telegram bot.
run_job	Run a job by its ID.
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, _post_init: Callable, _post_shutdown: Callable) -> None:
    """
    Initialize the Telegram client.

    Args:
        _post_init (callable): Function to call after the application is initialized.
        _post_shutdown (callable): Function to call after the application is shut down.

    """
    settings = get_settings()

    if settings.testing:
        self._base_url = "https://api.telegram.org/bot{token}/test"
        self._base_file_url = "https://api.telegram.org/file/bot{token}/test"
        self._testing = True

    # Set up the application with all the settings
    self._builder = Application.builder()
    self._builder.base_url(self._base_url)
    self._builder.base_file_url(self._base_file_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)

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

add_datasources

add_datasources(datasources: dict[str, DataSource]) -> None

Add data sources to the Telegram client.

Parameters:

Name	Type	Description	Default
datasources	dict[str, DataSource]	Dictionary of data source names and their corresponding callables.	required

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

def add_datasources(self, datasources: dict[str, DataSource]) -> None:
    """
    Add data sources to the Telegram client.

    Args:
        datasources (dict[str, DataSource]): Dictionary of data source names and their corresponding callables.

    """
    self.app.bot_data["datasources"] = datasources

add_default_handlers

add_default_handlers() -> None

Add default handlers to the Telegram client.

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

def add_default_handlers(self) -> None:
    """Add default handlers to the Telegram client."""
    settings = get_settings()
    with logger.catch(exception=TelegramError, level="ERROR", message="Failed to register default handlers"):
        if settings.responses.default_enabled:
            self.app.add_handler(MessageHandler(filters.TEXT, default))
        self.app.add_error_handler(error)

add_handlers

add_handlers(handlers: list[BaseHandler]) -> None

Add handlers to the Telegram client.

Parameters:

Name	Type	Description	Default
handlers	list[BaseHandler]	List of handlers to add.	required

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

def add_handlers(self, handlers: list[BaseHandler]) -> None:
    """
    Add handlers to the Telegram client.

    Args:
        handlers (list[BaseHandler]): List of handlers to add.

    """
    for handler in handlers:
        with logger.catch(exception=TelegramError, level="ERROR", message="Failed to register handler"):
            self.app.add_handler(handler)

add_jobs

add_jobs(
    jobs: list[
        tuple[
            Job,
            Callable[
                [CallbackContext], Coroutine[Any, Any, None]
            ],
        ]
    ],
) -> None

Add jobs to the Telegram client.

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

def add_jobs(self, jobs: list[tuple[Job, Callable[[CallbackContext], Coroutine[Any, Any, None]]]]) -> None:
    """Add jobs to the Telegram client."""
    if not get_settings().jobs.enabled:
        logger.debug("Jobs are disabled, skipping job registration")
        return

    logger.trace("Registering jobs...")
    self.app.job_queue.scheduler.remove_all_jobs()
    logger.trace("Removed all existing jobs")
    with Session(get_engine()) as session:
        for job, callback in jobs:
            session.add(job)
            with logger.catch(exception=TelegramError, level="ERROR", message="Failed to register job"):
                lg = logger.bind(job_id=job.id, action="/" + job.action.name, cron_expression=job.cron_expression)
                if not job.enabled:
                    lg.info("Disabled, skipping")
                    continue
                lg.trace("Registering job")
                self.app.job_queue.run_custom(
                    callback,
                    job_kwargs={
                        "trigger": CronTrigger.from_crontab(job.cron_expression),
                        "replace_existing": True,
                    },
                    data={
                        "args": job.args,
                        "per_user": job.per_user,
                        "users": [user.telegram_id for user in job.effective_users],
                    },
                    name=job.id,
                )
                lg.debug("Job registered")
    logger.debug("All jobs registered", jobs=len(jobs))

add_pages_handler

add_pages_handler() -> None

Add the pages handler to the Telegram client.

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

def add_pages_handler(self) -> None:
    """Add the pages handler to the Telegram client."""
    with logger.catch(exception=TelegramError, level="ERROR", message="Failed to register pages handler"):
        self.app.add_handler(CallbackQueryHandler(page_callback, pattern=rf"^{UUID4_REGEX.pattern}#[0-9]+$"))

reset_scopes async

reset_scopes() -> None

Reset the command scopes for the bot.

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

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

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

    This method clears all command scopes and sets the default commands.
    """
    if self._testing:
        logger.debug("Testing mode, skipping resetting scopes")
        return

    with logger.catch(exception=TelegramError, message="Failed to reset scopes"):
        await self.app.bot.delete_my_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)

run_job async

run_job(job_id: str) -> None

Run a job by its ID.

Parameters:

Name	Type	Description	Default
job_id	str	The ID of the job to run.	required

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

async def run_job(self, job_id: str) -> None:
    """
    Run a job by its ID.

    Args:
        job_id (str): The ID of the job to run.

    """
    job = self.app.job_queue.get_jobs_by_name(job_id)
    if not job:
        logger.warning(f"Job with ID {job_id} not found")
        return
    lg = logger.bind(job_id=job_id)
    lg.debug("Running job manually")
    await job[0].run(self.app)
    lg.debug("Job run completed")

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._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",
        ):
            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()