TelegramBot/classes/pyroclient.py

from typing import List, Union

from apscheduler.triggers.cron import CronTrigger
from libbot.pyrogram.classes import PyroClient as LibPyroClient
from pymongo import ASCENDING, GEOSPHERE, TEXT
from pyrogram.types import User

from classes.location import Location
from classes.pyrouser import PyroUser
from modules.database_api import col_locations
from modules.reminder import remind


class PyroClient(LibPyroClient):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        if self.scheduler is not None:
            self.scheduler.add_job(
                remind, CronTrigger.from_crontab("* * * * *"), args=(self,)
            )
        self.contexts = []

    async def start(self, **kwargs):
        await col_locations.create_index(
            [("id", ASCENDING)], name="location_id", unique=True
        )
        await col_locations.create_index(
            [("location", GEOSPHERE)],
            name="location_location",
        )
        await col_locations.create_index([("name", TEXT)], name="location_name")
        return await super().start(**kwargs)

    async def find_user(self, user: Union[int, User]) -> PyroUser:
        """Find User by it's ID or User object.

        ### Args:
            * user (`Union[int, User]`): ID or User object to extract ID from.

        ### Returns:
            * `PyroUser`: User in database representation.
        """

        return (
            await PyroUser.find(user)
            if isinstance(user, int)
            else await PyroUser.find(user.id, locale=user.language_code)
        )

    async def get_location(self, id: int) -> Location:
        """Get Location by it's ID.

        ### Args:
            * id (`int`): Location's ID. Defaults to `None`.

        ### Returns:
            * `Location`: Location from database as an object.
        """

        return await Location.get(id)

    async def list_locations(self) -> List[Location]:
        """Get all locations stored in database.

        ### Returns:
            * `List[Location]`: List of `Location` objects.
        """
        return [
            await Location.get(record["id"]) async for record in col_locations.find({})
        ]