Simple Bale Lib

.env

TOKEN=

lib.py

from dataclasses import asdict

import requests
from urllib.parse import urljoin
from utils import clear_none_value
from objects import Update, Message, LabeledPrice


class Bale:
    def __init__(self, token, update_handler, exception_handler, base_url='https://tapi.bale.ai/'):
        self.token = token
        self.base_url = base_url
        self.update_handler = update_handler
        self.exception_handler = exception_handler

    def _call(self, method, json=None, files=None):
        url = urljoin(self.base_url, f'/bot{self.token}/{method}')
        kwargs = {
            'json': json,
            'files': files,
        }
        try:
            response = requests.post(
                url=url,
                **clear_none_value(kwargs),
            )
            json_response = response.json()
            is_ok = json_response.get('ok')
            if is_ok:
                return json_response
            else:
                self.exception_handler(self, json_response)
        except Exception as e:
            print('Error on parse response')
        return

    def get_updates(self, offset=None, limit=None) -> list[Update]:
        json_data = self._call(
            method='getUpdates',
            json=clear_none_value({
                'offset': offset,
                'limit': limit,
            }),
        )
        result = []
        if not json_data:
            return []
        for update_dict in json_data['result']:
            result.append(Update.from_dict(update_dict))
        return result

    def send_message(self, chat_id: int, text: str, reply_to_message_id: int = None, reply_markup: dict = None):
        json_data = self._call(
            method='sendMessage',
            json=clear_none_value({
                'chat_id': chat_id,
                'text': text,
                'reply_to_message_id': reply_to_message_id,
                'reply_markup': reply_markup,
            }),
        )
        if json_data:
            return Message.from_dict(json_data['result'])

    def send_invoice(
            self,
            chat_id: int,
            title: str,
            description: str,
            payload: str,
            provider_token: str,
            prices: list[LabeledPrice],
    ):
        prices = [asdict(price) for price in prices]
        json_data = self._call(
            method='sendInvoice',
            json=clear_none_value({
                'chat_id': chat_id,
                'title': title,
                'description': description,
                'payload': payload,
                'provider_token': provider_token,
                'prices': prices,
            }),
        )
        if json_data:
            return Message.from_dict(json_data['result'])

    def send_video(
            self,
            chat_id: int,
            video: str | bytes,
            caption: str = None,
            reply_to_message_id: int = None,
            reply_markup: dict = None,
    ):
        if isinstance(video, str):
            extra_json = {
                'video': video,
            }
            extra_files = {}
        else:
            extra_json = {}
            extra_files = {
                'video': video,
            }
        json_data = self._call(
            method='sendVideo',
            json=clear_none_value({
                'chat_id': chat_id,
                'caption': caption,
                'reply_to_message_id': reply_to_message_id,
                'reply_markup': reply_markup,
                **extra_json,
            }),
            files={
                **extra_files,
            }
        )
        if json_data:
            return Message.from_dict(json_data['result'])

    def polling(self):
        last_update_id = None
        while True:
            updates = self.get_updates(last_update_id + 1 if last_update_id else last_update_id)
            if not updates:
                continue
            for update in updates:
                if last_update_id and (update.update_id <= last_update_id):
                    continue
                self.update_handler(self, update)
            last_update_id = updates[-1].update_id

main.py

from lib import Bale
from settings import settings
from objects import Update


def update_handler(bot: Bale, update: Update):
    bot.send_message(
        chat_id=update.message.chat.id,
        text='hi',
        reply_markup={
            'inline_keyboard': [
                [
                    {
                        'text': 'hi',
                        'callback_data': 'hi',
                    }
                ]
            ]
        }
    )


def exception_handler(bot: Bale, exception_obj):
    pass


if __name__ == '__main__':
    app = Bale(
        token=settings.TOKEN,
        update_handler=update_handler,
        exception_handler=exception_handler,
    )
    app.polling()

objects.py

from dataclasses import dataclass
from datetime import datetime
from typing import Optional


@dataclass
class Photo:
    small_file_id: str
    small_file_unique_id: str
    big_file_id: str
    big_file_unique_id: str

    @classmethod
    def from_dict(cls, dict_obj: dict) -> "Photo":
        dict_obj = dict_obj if dict_obj is not None else {}
        return cls(**dict_obj) if dict_obj else None


@dataclass
class User:
    id: int
    is_bot: bool
    first_name: str
    last_name: str
    username: str | None = None

    @classmethod
    def from_dict(cls, dict_obj: dict) -> "User":
        dict_obj = dict_obj if dict_obj is not None else {}
        return cls(**dict_obj) if dict_obj else None


@dataclass
class Chat:
    id: int
    type: str
    username: Optional[str] = None
    first_name: Optional[str] = None
    photo: Optional[Photo] = None

    @classmethod
    def from_dict(cls, dict_obj: dict) -> "Chat":
        dict_obj = dict_obj if dict_obj is not None else {}
        photo = dict_obj.pop('photo', None)
        return cls(
            photo=Photo.from_dict(photo) if photo else photo,
            **dict_obj,
        ) if dict_obj else None


@dataclass
class ForwardFromChat:
    id: int
    type: str
    title: str
    username: str
    photo: Photo

    @classmethod
    def from_dict(cls, dict_obj: dict) -> "ForwardFromChat":
        dict_obj = dict_obj if dict_obj is not None else {}
        photo = dict_obj.pop('photo', None) if dict_obj else None
        return cls(
            photo=Photo.from_dict(photo),
            **dict_obj
        ) if dict_obj else None


@dataclass
class Document:
    file_id: str
    file_unique_id: str
    file_name: str
    mime_type: str
    file_size: int

    @classmethod
    def from_dict(cls, dict_obj: dict) -> "Document":
        dict_obj = dict_obj if dict_obj is not None else {}
        return cls(**dict_obj) if dict_obj else None


@dataclass
class Video:
    file_id: str
    file_unique_id: str
    width: int
    height: int
    file_size: int
    thumb: Optional['Video'] = None
    mime_type: str | None = None
    duration: int | None = None

    @classmethod
    def from_dict(cls, dict_obj: dict) -> "Video":
        dict_obj = dict_obj if dict_obj is not None else {}
        return cls(**dict_obj) if dict_obj else None


@dataclass
class Message:
    message_id: int
    date: datetime
    from_user: User
    chat: Chat
    text: str | None = None
    caption: str | None = None
    document: Document | None = None
    video: Video | None = None
    forward_from: User | None = None
    forward_from_message_id: int | None = None
    forward_from_chat: ForwardFromChat | None = None

    @classmethod
    def from_dict(cls, dict_obj: dict) -> "Message":
        dict_obj = dict_obj if dict_obj is not None else {}
        date = dict_obj.pop('date', None)
        from_user = dict_obj.pop('from', None)
        chat = dict_obj.pop('chat', None)
        document = dict_obj.pop('document', None)
        video = dict_obj.pop('video', None)
        forward_from = dict_obj.pop('forward_from', None)
        forward_from_chat = dict_obj.pop('forward_from_chat', None)
        return cls(
            date=datetime.fromtimestamp(date) if date else None,
            from_user=User.from_dict(from_user),
            chat=Chat.from_dict(chat),
            document=Document.from_dict(document),
            video=Video.from_dict(video),
            forward_from=User.from_dict(forward_from),
            forward_from_chat=ForwardFromChat.from_dict(forward_from_chat),
            **dict_obj,
        ) if dict_obj else None


@dataclass
class Update:
    update_id: int
    message: Optional[Message]

    @classmethod
    def from_dict(cls, dict_obj: dict) -> "Update":
        dict_obj = dict_obj if dict_obj is not None else {}
        message = dict_obj.pop('message', None)
        return cls(
            message=Message.from_dict(message),
            **dict_obj,
        ) if dict_obj else None


@dataclass
class LabeledPrice:
    label: str
    amount: int

    @classmethod
    def from_dict(cls, dict_obj: dict) -> "LabeledPrice":
        return cls(**dict_obj) if dict_obj else None

requirements.txt

python-decouple==3.8
requests==2.31

settings.py

from decouple import config


class Setting:
    TOKEN = config('TOKEN')


settings = Setting()

utils.py

def clear_none_value(dict_obj: dict) -> dict:
    return {k: v for k, v in dict_obj.items() if v is not None}