garnet

About

garnet is a ridiculously simple library created mainly for managing your stateful telegram bots written with Telethon.

.. invisible-content .. _aiogram: https://github.com/aiogram/aiogram

How to install?

Although, garnet is garnet, it is named telegram-garnet on the PyPI, you'll have to tell that to pip.

pip install -U telegram-garnet

Let's dive in

.. code:: python

# export BOT_TOKEN, APP_ID, APP_HASH, SESSION_DSN env vars.
from garnet import ctx
from garnet.events import Router
from garnet.filters import State, text, group
from garnet.storages import DictStorage

router = Router()
UserStates = group.Group.from_iter(["echo"])  # declare users states

# register handler for "/start" commands for users with none yet set state
@router.message(text.commands("start"), State.entry)
async def entrypoint(event):
    await event.reply("You entered echo zone!\n/cancel to exit")
    fsm = ctx.CageCtx.get()
    await fsm.set_state(UserStates.echo)

# register handler for "/cancel" commands for users that have entered any state
@router.message(text.commands("cancel"), State.any)
async def cancel(event):
    await event.reply("Cancelled :)\n/start to restart")
    await ctx.CageCtx.get().set_state(None)

# handle any message from users with state=UserState.echo
@router.message(State.exact(UserStates.echo))
async def echo(event):
    await event.reply(event.text)

if __name__ == "__main__":
    from garnet.runner import run, launch
    launch("my-first-garnet-app", run(router, DictStorage()))

Key features

Filters

Basically Filter is a "lazy" callable which holds an optional single-parameter function. Filters are event naive and event aware. Filters are mutable, they can migrate from event-naive to event-aware in garnet.

Public methods

.is_event_naive -> bool
.call(e: T, /) -> Awaitable[bool]

Initializer ^^^^^^^^^^^

Filter(function[, event_builder])

Value of the parameter function must be function that takes exactly one argument with type Optional[Some] and returns bool either True or False.

Possible operations on Filter instances ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

(those are, primarily logical operators)

Binary """"""

& is a logical AND for two filters
| is a logical OR for two filters
^ is a logical XOR for two filters

Unary """""

~ is a logical NOT for a filter

Examples

.. code:: python3

from garnet import Filter, events

async def fun(_): ...

# example of event aware filter
Filter(fun, events.NewMessage)

# example of event-naive
Filter(fun)

By default Filter is event-naive, however when using with garnet::Router for handlers it may be changed.

Filters "from the box"

Text filters ^^^^^^^^^^^^

Operations on Filter((e: Some) -> bool); Some.raw_text or Some.text

Import """"""

from garnet.filters import text

Little journey """"""""""""""

text.Len is a special class for len(Some.raw_text ... "") operations. Supports logical comparison operations, such are ==, >, >=, <, <=
text.startswith(prefix: str, /) will evaluates to Some.raw_text.startswith(prefix)
text.commands(*cmds: str, prefixes="/", to_set=True) will evaluate to check if command is within cmd (ignores mentions, and works on Some.text)
text.match(rexpr: str, flags=0, /) will evaluate to re.compile(rexpr, flags).match(Some.raw_text)
text.between(*texts: str, to_set=True) will evaluate to Some.raw_text in texts
text.can_be_int(base=10) will evaluate to try{int(Some.raw_text);return True;}except(ValueError){return False;}
text.can_be_float() similarly to text.can_be_int but for floats.

State filters ^^^^^^^^^^^^^

Operations on users' states.

Import """"""

from garnet.filters import State

Little journey """"""""""""""

State.any will evaluate to match any state but not None
State.entry will evaluate to True if only current state is None
State.exact(state: GroupT | M | "*") when "*" is passed will use State.any, when states group is passed will check if current state is any states from the group, when state group member (M) passed will check if current state is exactly this state
State == {some} will call State.exact(state=some)

Note """"

State filter has effect on garnet.ctx.MCtx. And if you're not sure what are you doing try not to apply logical operators on State filters. Simply, don't do ~State.any or ~State.exact(...some...)

States declaration ^^^^^^^^^^^^^^^^^^

Import """"""

from garnet.filters import group

group.M (state group Member) """"""""""""""""""""""""""""

yes, "M" stands for member.

.next return the next M in the group or raise group.NoNext exception
.prev return the previous M in the group or raise group.NoPrev exception
.top return the top (head) M in the group or raise group.NoTop exception

group.Group """""""""""

Group of state members declared as a class (can be nested)

.first returns (M) the first declared member
.last returns (M) the last declared member

Note .first and .last are reserved "keywords" for state

Usage """""

.. code:: python

from garnet.filters import group, State

class Users(group.Group):
    ask_name = group.M()
    ask_age = group.M()

    class Pet(group.Group):
        ask_name = group.M()
        ask_age = group.M()

    class Hobby(group.Group):
        frequency = group.M()
        ask_if_popular = group.M()

# 💫 just imagine we already have router 💫

@router.default(State.exact(Users))  # will handle all states in "Users"
# --- some code ---
@router.default(State.exact(Users.Pet.ask_age))  # will handle only if current state is equal to "Users.Pet.ask_age"
# --- some code ---

Note """"

Think of groups as an immutable(not really...) linked list of connected group members As you can see in the example above we use nested states groups. One thing about about M.[next/prev/top]. We can go to Users.Pet.ask_name from Users.ask_age using Users.ask_age.next, but not backwards as someone could expect with Users.Pet.ask_name.prev (will actually raise NoPrev) Nested group members do not know anything about upper members, but they have "owners" which have access to their parent groups and in order to access parent of owner of x = Users.Pet.ask_name, we would use x.owner

Callback query (QueryBaker) ^^^^^^^^^^^^^^^^^^^^^^^^^^^

Operations on callback queries. Baker is a callback_data string generator/parser/validator. garnet.ctx::Query has context value which is set after every successful validation.

Import """"""

from garnet.filters import QueryBaker

Little journey """"""""""""""

(prefix:str, /, *args:str, [ignore:Iterable[QItem]=(),][sep:str="\n",][maxlen:int=64]) initializer function, if you want to have custom types in QueryDict
.filter(extend_ignore:Iterable[str]=(), /, **config) will make sure user given callback data is valid by given config.
.get_checked(**non_ignored:Any) will return a string based on passed passed args

Usage """""

.. code:: python

from garnet.filters import QueryBaker

qb = QueryBaker(
    "v",  # set v string as identity(prefix) for our baker
    ("id", uuid.UUID),  # make uuid.UUID a factory for id arg
    "act",
    ignore=("id",),  # mark id arg as `optional`
    sep=":",  # set a separator for arg values, better not change
    maxlen=64,  # get_checked will check the length of generated callback and tell you if it's more than maxlen
)
# create v:{id}:{act} pattern

qb.filter(act="apply")
# will be a filter to match queries like "v:(.*):apply"

qb.get_checked(id="51b3f442-a9f6-4dcc-918e-1f08b1189386", act="clear")
# will produce a "safe" string pattern v:51b3f442-a9f6-4dcc-918e-1f08b1189386:clear

# Where you may want to use .get_checked
Button.inline(..., data=qb.get_checked(id=<...>, act="mpa"))

Note """"

Don't use separator string inside your arg values.

To reuse validated data from filter, use Query (validated dict)_

Routers

Router (routing table) is a collection of handlers.

Public methods

Those consist mainly from decorators.

Initializer ^^^^^^^^^^^

Router(default_event=None, *filters)

default_event default event builder for router
*filters router filters, in order to get into handlers, event should pass these filters.

Decorators ^^^^^^^^^^

Depending on event_builder of a decorator, filters inherit that event builder mutating themselves.

.default(*filters) event builder is default Router(this, ...), should not be None, must implement telethon.common::EventBuilder
.message(*filters) shortcut decorator for event builder garnet.events::NewMessage
.callback_query(*filters) shortcut decorator for event builder garnet.events::CallbackQuery
.chat_action(*filters) shortcut decorator for event builder garnet.events::ChatAction
.message_edited(*filters) shortcut decorator for event builder garnet.events::MessageEdited
.on(event_builder, /, *filters) pass any event builder (preferably from garnet.events::*)
.use() use this decorator for intermediates that are called after filters

etc. ^^^^

.add_use(intermediate, /) register an intermediate which will be called after filters for handlers
.register(handler, filters, event_builder) register handler with binding filters and event_builder to it.
``.include(ro

Garnet

Install / Use

README

Filters

Public methods

Examples

Filters "from the box"

Routers

Public methods