Garnet - cool add-on for Telethon
Project description
Install:
pip install telegram-garnet
- Dependencies:
telethon - main dependency telethon
- Extras:
🌚 🌝 FSM-Storage types
File - json storage, the main idea behind JSON storage is a custom reload of file and usage of memory session for writing, so the data in json storage not always actual
Memory - powerful in-memory map<str> based storage, only thing - not persistent
Redis - (requires aioredis) - redis is fast key-value storage, if you’re using your data is durable and persistent
Pomegranate implements updates dispatching and checks callback’s filters wrapping all callbacks into garnet::Callback object
😋 Filters
Filter object is the essential part of Pomegranate, the do state checking and other stuff.
Basically, it’s func from MyEventBuilder(func=lambda self: <bool>) but a way more complicated and not stored in EventBuilder, it’s stored in callback object
Useful filters
📨 Messages
Following examples will include pattern
from garnet import MessageText, TelegramClient
bot = TelegramClient.from_env().start_as_bot()
# // our code here //
bot.run_until_disconnected()
# handling /start and /help
@bot.on(MessageText.commands("help", "start"))
async def cmd_handler(message: custom.Message):
await message.reply("Hey there!")
# handling exact words
my_secret = "abcxyz"
@bot.on(MessageText.exact(my_secret))
async def secret_handler(message: custom.Message):
await message.reply("Secret entered correctly! Welcome!")
- MessageText or text(from garnet import text) includes following comparisons all returning <bool>
.exact(x) -> event.raw_text == x
.commands(*x) -> event.raw_text.split()[0][1:] in x
.match(x) -> re.compile(x).match(event.raw_text)
.between(*x) -> event.raw_text in x
.isdigit() -> (event.raw_text or "").isdigit()
.startswith(x) -> event.raw_text.startswith(x)
👀 CurrentState class [from garnet import CurrentState]
Once great minds decided that state checking will be in filters without adding state as handler decorator parameter and further storing state in callback.(arg) CurrentState class methods return Filter. There are two problems that Filter object really solves, Filter’s function can be any kind of callable(async,sync), filters also have a flag requires_context, FSMProxy is passed if true
See FSM example to understand how CurrentState works
- Includes following methods all returning <bool>
.exact(x) or CurrentState == x -> await context.get_state() == x
CurrentState == [x, y, z] -> await context.get_state() in [x, y, z]
CurrentState == all or CurrentState == any -> await context.get_state() is not None
🦔 Custom Filter
If you want to write your own filter, do it.
from garnet import Filter, FSMProxy
async def myFunc(event): ...
async def myFuncContextRequires(event, context: FSMProxy): ...
def normal_func(event): ...
@bot.on(Filter(normal_func), Filter(myFunc), Filter(myFuncContextRequires, requires_context=True))
async def handler(event, context: FSMProxy): ...
# same as
@bot.on(normal_func, myFunc, Filter(myFuncContextRequires, requires_context=True))
async def handler(event): ...
So the handler can take strict context argument and also ignore it
On start|finish
garnet::TelegramClient contains three lists on_start on_background and on_finish, their instance is PseudoFrozenList which freezes at calling .run_until_disconnected PseudoFrozenList has three main methods:
.append(*items) .remove(*items) .freeze() .__call__(func) # for shiny decorator
items in case of TelegramClient means unpacked container of async-defined functions taking on position arguments
Usage example:
# my_module.py
class MyPostgresDatabase:
...
async def close_pool(self, bot): await self.pool.safe_close()
async def open_pool(self, bot): await self.pool.open_conn_pool()
# garnethon.py
from garnet import TelegramClient
from my_module import MyPostgresDatabase
db = MyPostgresDatabase()
bot = TelegramClient.from_env().start_as_bot()
bot.on_start.append(db.open_pool)
bot.on_finish.append(db.close_pool)
...
@bot.on_background
async def xyz(cl: TelegramClient):
while True:
...
bot.run_until_connected()
📦 Router and Migrating to garnet using Router
Think of router as just a dummy container of handlers(callbacks)
garnet::router::Router may be helpful if you have telethon’s event.register registered handlers. One thing: Router, I believe, is correct and more obvious way of registering event handlers. Example:
Migrate from telethon to garnet, also use for bot.on cases(soon better example)
# my_handlers.py
# telethon register(bad) will raise Warning in garnet
from telethon import events
@events.register(event_type)
async def handler(event): ...
# garnet's telethon-like router
from garnet.router import TelethonRouter
router = TelethonRouter()
@router(event_type)
async def handler(event): ...
The advantage of routers is evidence of registering handlers when you have module-separated handlers. events.register was doing well, but blindly importing modules to register handlers and don’t use them(modules) doesn’t seem like a good idea.
Example of registering router in bot application
# handlers/messages.py
from garnet.router import Router
router = Router()
@router()
async def handler(event): ...
# handlers/cb_query.py
from garnet.events import CallbackQuery
from garnet.router import Router
router = Router()
@router(event=CallbackQuery())
async def handler(event): ...
# entry.py ()
from garnet import TelegramClient
from handlers import messages, cb_query
tg = TelegramClient.from_env().start_as_bot()
tg.bind_routers(messages, cb_query)
...
TelethonRouter and Router both have following remarkable methods:
.message_handler(*filters) .callback_query_handler(*filters) .chat_action_handler(*filters) .message_edited_handler(*filters) .album_handler(*filters)
🍬 Context magic
One of the sweetest parts of garnet. Using contextvars we reach incredibly beautiful code :D this is not FSMContext don’t confuse with context magic provided by contextvars
As an example, bot that doesn’t requires TelegramClient to answer messages directly.
from garnet.functions.messages import reply, message, respond
@bot.message_handler()
async def handler():
# message() - function to get current Message event
await message().respond("ok")
await message().reply("ok")
# the same result, but shortcuts
await respond("ok")
await reply("Ok")
What’s more ❓
Class-based handlers are also can be implemented with garnet conveniently. Use your imagination and garnet::callbacks::base::Callback as a parent class
Awesome bitwise operation supported filters(I highly recommend to use them):
# & (conjunction), | (disjunction), ~ (inversion), ^ (exclusive disjunction) # also: ==, != (idk why) @bot.on(MessageText.exact(".") | MessageText.exact(".."))
Len attribute in MessageText which has cmp methods:
@bot.on((MessageText.Len <= 14) | (MessageText.Len >= 88))
Using client = TelegramClient.start assignment and start client on the fly, make annotation or typing.cast to have better hints.
About
You can find me in tg by @martin_winks and yeah I receive donates as well as all contributors do(support lonamiwebs and JRootJunior).
🤗 Credits
Finite-state machine was ported from cool BotAPI library ‘aiogram’, special thanks to Alex
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for telegram_garnet-0.2.3-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 46e8fbe405b988bce0120185c3b5be562f91d1df9e4afcf1eda514bd9b6eaa02 |
|
MD5 | 5d6090470864d9d611a4572917748471 |
|
BLAKE2b-256 | 85cc7ec352952bd96757334ff8cdb6291f6541e5c4fedf39a0234abb1dc69365 |