kamihi.bot

Bot module for Kamihi.

This module provides the primary interface for the Kamihi framework, allowing for the creation and management of Telegram bots.

License

MIT

Modules:

Name	Description
action	Action helper class.
bot	Bot module for Kamihi.

Classes:

Name	Description
Action	Action class for Kamihi bot.
Bot	Bot class for Kamihi.

Action

Action(
    name: str,
    commands: list[str],
    description: str,
    func: Callable,
    datasources: dict[str, DataSource] = None,
)

Action class for Kamihi bot.

This class provides helpers for defining actions, their commands and their handlers.

Attributes:

Name	Type	Description
name	str	The name of the action.
commands	list[str]	List of commands associated.
description	str	Description of the action.

Initialize the Action class.

Parameters:

Name	Type	Description	Default
name	str	The name of the action.	required
commands	list[str]	List of commands associated.	required
description	str	Description of the action.	required
func	Callable	The function to be executed when the action is called.	required
datasources	dict[str, DataSource]	Dictionary of data sources available for the action.	None

Methods:

Name	Description
clean_up	Clean up the action from the database.
run_scheduled	Execute the action in a job context.

Source code in src/kamihi/bot/action.py

def __init__(
    self,
    name: str,
    commands: list[str],
    description: str,
    func: Callable,
    datasources: dict[str, DataSource] = None,
) -> None:
    """
    Initialize the Action class.

    Args:
        name (str): The name of the action.
        commands (list[str]): List of commands associated.
        description (str): Description of the action.
        func (Callable): The function to be executed when the action is called.
        datasources (dict[str, DataSource]): Dictionary of data sources available for the action.

    """
    self.name = name
    self.commands = commands
    self.description = description

    self._func = func
    self._logger = logger.bind(action=self.name)

    self._datasources = datasources or {}

    self._files = Environment(
        loader=FileSystemLoader(self._folder_path),
        autoescape=select_autoescape(default_for_string=False),
    )

    self._validate_commands()
    self._validate_function()
    self._validate_requests()

    self._save_to_db()

    self._logger.debug("Successfully registered")

handler property

handler: AuthHandler | ConversationHandler

Construct a CommandHandler for the action.

jobs property

jobs: list[
    tuple[
        Job,
        Callable[
            [CallbackContext], Coroutine[Any, Any, None]
        ],
    ]
]

Return a list of jobs associated with the action.

users property

users: Sequence[BaseUser]

Return a list of telegram IDs of users authorized to use the action.

clean_up classmethod

clean_up(keep: list[str]) -> None

Clean up the action from the database.

Source code in src/kamihi/bot/action.py

@classmethod
def clean_up(cls, keep: list[str]) -> None:
    """Clean up the action from the database."""
    with Session(get_engine()) as session:
        statement = select(RegisteredAction).where(RegisteredAction.name.not_in(keep))
        actions = session.execute(statement).scalars().all()
        for action in actions:
            session.delete(action)
        session.commit()

run_scheduled async

run_scheduled(context: CallbackContext) -> None

Execute the action in a job context.

Source code in src/kamihi/bot/action.py

async def run_scheduled(self, context: CallbackContext) -> None:
    """Execute the action in a job context."""
    per_user = context.job.data.get("per_user", False)

    self._logger.debug("Executing scheduled action", per_user=per_user)

    if per_user:
        for telegram_id in context.job.data.get("users", []):
            async with self._db_session(context) as session:
                context.job.data["user"] = telegram_id
                pos_args, keyword_args = await self._fill_parameters(session, context)
                result: Any = await self._func(*pos_args, **keyword_args)
                await send(result, telegram_id, context)
    else:
        async with self._db_session(context) as session:
            pos_args, keyword_args = await self._fill_parameters(session, context)
            result: Any = await self._func(*pos_args, **keyword_args)
            for telegram_id in context.job.data.get("users", []):
                await send(result, telegram_id, context)

    self._logger.debug("Finished scheduled execution")

Bot

Bot()

Bot class for Kamihi.

The framework already provides a bot instance, which can be accessed using the bot variable. This instance is already configured with default settings and can be used to start the bot. The managed instance is preferable to using the Bot class directly, as it ensures that the bot is properly configured and managed by the framework.

Initialize the Bot class.

Methods:

Name	Description
action	Register an action with the bot.
start	Start the bot.

Source code in src/kamihi/bot/bot.py

def __init__(self) -> None:
    """Initialize the Bot class."""
    # Loads the datasources
    settings = get_settings()

    for datasource_config in settings.datasources:
        datasource_class = DataSource.get_datasource_class(datasource_config.type)
        try:
            self.datasources[datasource_config.name] = datasource_class(datasource_config)
        except ImportError as e:
            msg = (
                f"Failed to initialize data source '{datasource_config.name}' "
                f"of type '{datasource_config.type}' because of missing dependencies."
            )
            raise ImportError(msg) from e

        logger.trace("Initialized", datasource=datasource_config.name, type=datasource_config.type)

action

action(
    *commands: str, description: str = None
) -> partial[Action]

Register an action with the bot.

This method overloads the bot.action method so the decorator can be used with or without parentheses.

Parameters:

Name	Type	Description	Default
*commands	str	A list of command names. If not provided, the function name will be used.	()
description	str	A description of the action. This will be used in the help message.	None

Returns:

Name	Type	Description
Callable	partial[Action]	The wrapped function.

Source code in src/kamihi/bot/bot.py

@dispatch([str])
def action(self, *commands: str, description: str = None) -> partial[Action]:
    """
    Register an action with the bot.

    This method overloads the `bot.action` method so the decorator can be used
    with or without parentheses.

    Args:
        *commands: A list of command names. If not provided, the function name will be used.
        description: A description of the action. This will be used in the help message.

    Returns:
        Callable: The wrapped function.

    """
    return functools.partial(self.action, *commands, description=description)

start

start() -> None

Start the bot.

Source code in src/kamihi/bot/bot.py

def start(self) -> None:
    """Start the bot."""
    # Cleans up the database of actions that are not present in code
    Action.clean_up([action.name for action in self._actions])
    logger.debug("Removed actions not present in code from database")

    # Warns the user if there are no valid actions registered
    if not self._actions:
        logger.warning("No valid actions were registered. The bot will not respond to any commands.")

    # Cleans up the database of old pages
    Pages.clean_up(get_settings().db.pages_expiration_days)
    logger.trace("Cleaned up old pages")

    # Loads the Telegram client
    self._client = TelegramClient(self._post_init, self._post_shutdown)
    self._client.add_datasources(self.datasources)
    self._client.add_handlers(self._handlers)
    self._load_jobs()
    self._client.add_default_handlers()
    self._client.add_pages_handler()
    logger.trace("Initialized Telegram client")

    # Loads the web server
    self._web = KamihiWeb(
        {
            "after_create": [self._set_scopes, self._load_jobs],
            "after_edit": [self._set_scopes, self._load_jobs],
            "after_delete": [self._set_scopes, self._load_jobs],
            "run_job": [self._run_job],
        },
    )
    logger.trace("Initialized web server")
    self._web.start()

    # Runs the client
    self._client.run()