timoniq / telegrinder Goto Github PK

from telegrinder import LoopWrapper

async def create_tables() -> None:
    ...

lw = LoopWrapper()
lw.lifespan.on_startup(create_tables())
lw.run_event_loop()

Option = Some[Value] | Nothing

from telegrinder.option import Option, Nothing
from telegrinder.model import Model

class User(Model):
    id: int
    is_bot: bool
    first_name: str
    last_name: Option[str] = Nothing

def repr_user(user: User) -> str:
    # returns first_name + last_name
    return user.first_name + user.last_name.map(lambda value: " " + value).unwrap_or("")

class MessageView(ABCView[MessageCute]):
    async def check(self, event: Update) -> bool:
        return bool(event.message)

    async def process(self, event: Update, api: ABCAPI):
         # event.message.unwrap() is guaranteed to return variant Some after check method, which is good for logic and linter
         msg = MessageCute(**event.message.unwrap().to_dict(), api=api)
         return await process_inner(msg, event, self.middlewares, self.handlers)

@node
def/async def db_connection() -> Connection:
    conn = db.connect()
    yield conn
    conn.close()

from telegrinder import API, Telegrinder, Token, Message
from telegrinder.rules import MessageRule, Markup

api = API(Token.from_env())
bot = Telegrinder(api)


class WithReply(MessageRule):
    async def check(self, message: Message, ctx: dict) -> bool:
        ctx.clear()  # accidental context deletion
        return message.reply_to_message is not None

        
@bot.on.message(Markup(["/give <item> <count:int>", "/give <item>"]), WithReply())
async def give_item(message: Message, item: str, count: int = 1): 
    # an exception will be raised because the ctx was
    # empty and there is nothing pass to the handler
    ...

from telegrinder import InlineKeyboardEnum, KeyboardEnum, InlineButton, Button


class FoodKeyboard(InlineKeyboardEnum):
    PRETZEL = InlineButton("pretzel", callback_data="food/pretzel")
    PIZZA = InlineButton("pizza", callback_data="food/pizza")
    PASTA = InlineButton("pasta", callback_data="food/pasta")
    RICE = InlineButton("rice", callback_data="food/rice")
    DUMPLINGS = InlineButton("dumplings", callback_data="food/dumplings")


class MenuKeyboard(KeyboardEnum):
    START = Button("start")
    EAT = Button("eat")
    SEND_CONTACT = Button("send_contact", request_contact=True)


food_kb_markup = FoodKeyboard.get_markup()
menu_kb_markup = MenuKeyboard.get_markup()


@bot.on.callback_query(FoodKeyboard.PRETZEL)
...


@bot.on.message(MenuKeyboard.EAT)
...

InlineButton("Dummy button", callback_data={"action": "buy", "quantity": 3})

class MySet(RuleEnum):
    CANCEL = Text("/cancel")
    USERNAME = Mention() | Markup("t.me/<username>")

@dp.message(MySet())

class Dispatch(ABCDispatch):
    # ...
    message = MessageView()
    views = {"message"}

import dataclasses

import typing_extensions as typing

from telegrinder.bot.dispatch.view import ABCView, MessageView

MessageViewT = typing.TypeVar("MessageViewT", bound=ABCView, default=MessageView)


@dataclasses.dataclass(kw_only=True)
class Views(typing.Generic[MessageViewT]):
    message: MessageViewT = dataclasses.field(default_factory=lambda: MessageView())


class Dispatch(ABCDispatch, Views[MessageViewT]):
    ...


dp = Dispatch()
dp.message # default instance 'MessageView'
dp = Dispatch(message=CustomMessageView())
dp.message # custom instance 'CustomMessageView'

bot = Telegrinder(...)
await bot.dispatch.feed(update, api)

feed(update, api, context={"shop_id": shop_id})
# or
bot.dependency["shop_id"] = Value(shop_id)  
# ^ value wrapper is needed to separate values from async getters

kitten = CachedMedia(photo="assets/kitten.jpg")

async def runtime(...):
    await api.send_photo(message.chat.id, caption="you lol", photo=await kitten.get())

message = await wm.wait(view, event, Rule())
message, ctx = await wm.wait(view, event, Rule(), return_ctx=True)

class Rational(MessageRule, composite=True, requires=[HasText()]):
    nom: int
    denom: int

    async def check(self, message: Message) -> Result["Rational", str]:
        rational = parse_rational(message.text, raise_error=False)
        if rational is None:
            return Error("Text is not a rational number")
        return Ok(
            self.context(rational.nominator, rational.denominator)
        )

class Text(MessageRule):
    async def check(self, message: Message) -> Result[str, str]:
        if not message.text:
            return Error("Message has no text")
        return Ok(message.text)

class Rational(ABCRule, composite=True, adaptor=Text):
    async def check(self, text: str) -> Result["Rational", str]:
        ...

from telegrinder.rules import CallbackDataMap

@bot.on.callback_query(CallbackDataMap({"count": int}))
async def cb_handler(cb: CallbackQuery, count: int):
    await cb.answer(str(count))
 
bot.run_forever()

from envparse import Env
from telegrinder import Telegrinder, API, Token
from telegrinder import logger
from telegrinder.client import AiohttpClient
import asyncio
from app.handlers import dps

api = API(Token.from_env())
bot = Telegrinder(api)
env = Env()
logger.set_level(env.str("LOGGING"))
SUPPLIERS_URL_API = env.str("SUPPLIERS_URL_API")
DISCOUNT_URL_API = env.str("DISCOUNT_URL_API")


@bot.loop_wrapper.interval(minutes=10)
async def update():
    client = AiohttpClient()
    headers = {}
    response = await client.request_json(
        SUPPLIERS_URL_API + "/api/v3/orders/new", "GET", headers=headers
    )

    # other things


if __name__ == "__main__":
    for dp in dps:
        bot.dispatch.load(dp)
    bot.run_forever()

api = API(token=token, prase_mode=ParseMode.HTML)

async def handler(m):
    await m.answer(
        HTMLFormatter(bold(italic("bold italic text!"))),
    )

# instead of

async def handler(m):
  await m.answer(
      HTMLFormatter(bold(italic("bold italic text!"))),
      parse_mode=HTMLFormatter.PARSE_MODE,
  )

class WaiterMachine:
    def __init__(self, *, max_storage_size: int = 1000, base_task_lifetime: float | timedelta | None = None) -> None:
        self.max_storage_size = max_storage_size
        self.base_task_lifetime = base_task_lifetime
        self.storage: Storage = {}

_: KW_ONLY = ...

class Text(ScalarNode, str):
    @composer
    async def compose_message(cls, message: MessageNode) -> typing.Self:
        ...
        return cls(message.text.unwrap())
    
    @composer
    async def compose_caption(cls, attachment: Attachment) -> typing.Self:
        # works with attachments with caption
        ...
        return cls(attachment.caption.unwrap())