gramix 0.1.9

A fast, fully-typed Python framework for building Telegram bots.

gramix

A fast, clean, fully-typed Python framework for building Telegram bots. Supports synchronous and asynchronous execution, finite state machines, middleware, inline keyboards, webhooks, and file handling — with zero boilerplate.

⭐ GitHub Repository

---

Table of Contents

• Installation
• Quick Start
• Core Concepts
• Examples
  • Commands & Text Filters
  • Reply Keyboards
  • Inline Keyboards & Callbacks
  • Finite State Machine (FSM)
  • SQLite FSM Storage
  • Media Handling
  • Sending Media
  • File Download
  • Middleware
  • Inline Queries
  • Chat Member Events
  • Polls & Quiz
  • Location & Venue
  • Payments
  • Parse Mode & HTML Formatting
  • Telegram Games
  • Rate Limiting
  • Async Mode
  • Webhook Mode
  • Lifecycle Hooks
• API Reference
• License

---

Installation

pip install gramix

Optional extras for webhook support:

pip install gramix[aiohttp]   # aiohttp webhook backend
pip install gramix[fastapi]   # FastAPI + uvicorn webhook backend

Requirements: Python 3.10+

---

Quick Start

Create a .env file:

BOT_TOKEN=your_token_here

Create bot.py:

from gramix import Bot, Dispatcher, Router, load_env

load_env()

bot = Bot()
dp  = Dispatcher(bot)
rt  = Router()
dp.include(rt)

@rt.message("/start")
def on_start(msg):
    msg.answer(f"Hello, {msg.from_user.full_name}!")

dp.run()

python bot.py

---

Core Concepts

Bot handles all direct Telegram API calls. Pass parse_mode=ParseMode.HTML once at initialization and every subsequent answer(), reply(), edit(), send_photo(), etc. will inherit it automatically — no need to repeat it per call.

Dispatcher drives the polling or webhook loop, dispatches incoming updates to routers, and manages middleware and lifecycle hooks.

Router declares handlers for messages, callbacks, inline queries, and FSM states. Multiple routers can be included in one dispatcher.

F is a filter shortcut object for common conditions: F.photo, F.document, F.sticker, F.voice, F.text, F.reply, F.forward, F.private, F.group, F.supergroup, F.channel.

State / Step define FSM flows. Steps are declared as class attributes and traversed with ctx.next(), ctx.prev(), or ctx.finish().

---

Examples

Commands & Text Filters

from gramix import Bot, Dispatcher, Router, load_env, F

load_env()
bot = Bot()
dp  = Dispatcher(bot)
rt  = Router()
dp.include(rt)

@rt.message("/start")
def on_start(msg):
    msg.answer("Welcome! Send me anything.")

@rt.message("/help")
def on_help(msg):
    msg.answer("Commands: /start, /help, /about")

# Exact text match (case-insensitive by default)
@rt.message("ping")
def on_ping(msg):
    msg.answer("pong")

# Regex filter — matches any 4-digit number
@rt.message(regex=r"^\d{4}$")
def on_four_digits(msg):
    msg.answer(f"You sent a 4-digit number: {msg.text}")

# Catch all text messages
@rt.message(F.text)
def on_text(msg):
    msg.answer(f"You said: {msg.text}")

dp.run()

---

Reply Keyboards

from gramix import Reply, RemoveKeyboard

@rt.message("/start")
def on_start(msg):
    kb = (
        Reply(resize=True)
        .button("📋 Menu")
        .button("ℹ️ Info")
        .row()
        .button("⚙️ Settings")
    )
    msg.answer("Choose an option:", keyboard=kb)

@rt.message("/remove")
def on_remove(msg):
    msg.answer("Keyboard removed.", keyboard=RemoveKeyboard())

---

Inline Keyboards & Callbacks

from gramix import Inline

@rt.message("/vote")
def on_vote(msg):
    kb = (
        Inline()
        .button("👍 Like", callback="vote:like")
        .button("👎 Dislike", callback="vote:dislike")
        .row()
        .button("🔗 Source", url="https://github.com/riokzyofficial-debug/gramix")
    )
    msg.answer("Cast your vote:", keyboard=kb)

@rt.callback("vote:like", "vote:dislike")
def on_vote_result(cb):
    label = "👍 Like" if cb.data == "vote:like" else "👎 Dislike"
    cb.answer(f"You chose {label}", show_alert=True)
    cb.message.edit(f"You voted: {label}")

# Handle callbacks by prefix
@rt.callback(prefix="item:")
def on_item(cb):
    item_id = cb.data.split(":")[1]
    cb.answer(f"Selected item #{item_id}")
    cb.message.edit(f"You picked item #{item_id}.")

---

Finite State Machine (FSM)

from gramix import State, Step, MemoryStorage, Router, RemoveKeyboard

rt = Router(storage=MemoryStorage())

class Registration(State):
    name = Step()
    age  = Step()
    city = Step()

@rt.message("/register")
def on_register(msg):
    ctx = rt.fsm.get(msg.from_user.id)
    ctx.set(Registration.name)
    msg.answer("What is your name?", keyboard=RemoveKeyboard())

@rt.state(Registration.name)
def get_name(msg, ctx):
    ctx.data["name"] = msg.text
    ctx.next()
    msg.answer("How old are you?")

@rt.state(Registration.age)
def get_age(msg, ctx):
    if not msg.text.isdigit():
        msg.answer("Please enter a valid number.")
        return
    ctx.data["age"] = int(msg.text)
    ctx.next()
    msg.answer("Which city are you from?")

@rt.state(Registration.city)
def get_city(msg, ctx):
    ctx.data["city"] = msg.text
    ctx.finish()
    msg.answer(
        f"Done! Name: {ctx.data['name']}, "
        f"Age: {ctx.data['age']}, City: {ctx.data['city']}"
    )

---

SQLite FSM Storage

Use SQLiteStorage to persist FSM state across bot restarts:

from gramix import SQLiteStorage, Router

rt = Router(storage=SQLiteStorage("state.db"))

# All @rt.state() handlers work identically — storage is transparent.
# SQLiteStorage also supports context manager usage:
with SQLiteStorage("state.db") as storage:
    rt = Router(storage=storage)

---

Media Handling

from gramix import F

@rt.message(F.photo)
def on_photo(msg):
    largest = msg.photo[-1]
    msg.answer(f"Photo: {largest.width}×{largest.height}px")

@rt.message(F.document)
def on_document(msg):
    msg.answer(f"Document: {msg.document.file_name} ({msg.document.file_size} bytes)")

@rt.message(F.voice)
def on_voice(msg):
    msg.answer(f"Voice message: {msg.voice.duration}s")

---

Sending Media

@rt.message("/photo")
def on_send_photo(msg):
    msg.reply_photo("https://picsum.photos/800/600", caption="A random photo")

@rt.message("/video")
def on_send_video(msg):
    msg.reply_video("https://example.com/sample.mp4", caption="Sample video")

@rt.message("/document")
def on_send_document(msg):
    msg.reply_document("<file_id>", caption="Here is your file")

---

File Download

@rt.message(F.document)
def on_document(msg):
    file_bytes = bot.download_file(msg.document.file_id)
    with open(msg.document.file_name, "wb") as f:
        f.write(file_bytes)
    msg.answer(f"Saved: {msg.document.file_name}")

---

Middleware

import time

@dp.middleware
def timing_middleware(msg, call_next):
    start = time.monotonic()
    call_next()
    elapsed = time.monotonic() - start
    print(f"[{msg.from_user.id}] handled in {elapsed:.3f}s")

@dp.middleware
def auth_middleware(msg, call_next):
    ALLOWED = {123456789, 987654321}
    if msg.from_user.id not in ALLOWED:
        msg.answer("Access denied.")
        return
    call_next()

---

Inline Queries

from gramix import InlineQueryResultArticle

@rt.inline()
def on_inline(query):
    results = [
        InlineQueryResultArticle(
            id="1",
            title="gramix",
            message_text="gramix — Telegram bot framework: https://pypi.org/project/gramix",
            description="Python Telegram bot framework",
        ),
    ]
    query.answer(results, cache_time=10)

---

Chat Member Events

@rt.chat_member()
def on_member_update(update):
    if update.joined:
        print(f"{update.user.full_name} joined {update.chat.display_name}")
    elif update.left:
        print(f"{update.user.full_name} left {update.chat.display_name}")

---

Polls & Quiz

@rt.message("/poll")
def send_poll(msg):
    bot.send_poll(
        chat_id=msg.chat.id,
        question="What is your favourite language?",
        options=["Python", "TypeScript", "Rust", "Go"],
        is_anonymous=True,
    )

@rt.poll_answer()
def on_vote(answer):
    print(f"{answer.user.full_name} chose options {answer.option_ids}")

---

Location & Venue

@rt.message("/location")
def cmd_location(msg):
    bot.send_location(chat_id=msg.chat.id, latitude=55.7558, longitude=37.6173)

@rt.message(F.location)
def on_location(msg):
    loc = msg.location
    msg.answer(f"Received point: {loc.latitude}, {loc.longitude}")

---

Payments

from gramix import LabeledPrice

@rt.message("/buy")
def cmd_buy(msg):
    bot.send_invoice(
        chat_id=msg.chat.id,
        title="Premium доступ",
        description="30 дней Premium.",
        payload="premium_30d",
        provider_token="YOUR_PROVIDER_TOKEN",
        currency="RUB",
        prices=[LabeledPrice(label="Premium", amount=29900)],
    )

@rt.pre_checkout_query()
def on_pre_checkout(query):
    bot.answer_pre_checkout_query(query.id, ok=True)

@rt.successful_payment()
def on_payment(msg):
    p = msg.successful_payment
    msg.answer(f"Оплата прошла! {p.amount_decimal} {p.currency}")

---

Parse Mode & HTML Formatting

from gramix import Bot, ParseMode

bot = Bot(parse_mode=ParseMode.HTML)

@rt.message("/start")
def on_start(msg):
    msg.answer(
        "<b>Bold</b>, <i>italic</i>, <code>inline code</code>\n"
        f"Hello, <b>{msg.from_user.full_name}</b>!"
    )

---

Telegram Games

from gramix import Inline

@rt.message("/game")
def on_game(msg):
    kb = Inline().button("🎮 Play", callback="game_play")
    bot.send_game(chat_id=msg.chat.id, game_short_name="mygame", keyboard=kb)

@rt.game_callback()
def on_game_play(cb):
    cb.answer(url=f"https://yourdomain.com/game?user={cb.from_user.id}")

@rt.message("/scores")
def on_scores(msg):
    scores = bot.get_game_high_scores(
        user_id=msg.from_user.id,
        chat_id=msg.chat.id,
        message_id=msg.reply_to_message.message_id,
    )
    for entry in scores:
        msg.answer(f"#{entry.position} {entry.user.full_name}: {entry.score}")

---

Rate Limiting

from gramix import ThrottlingMiddleware

# Silent drop — allow one message per second per user
dp.middleware(ThrottlingMiddleware(rate=1.0))

# With a custom response when throttled
def on_throttle(msg):
    msg.answer("⚠️ Too many requests. Please wait a moment.")

dp.middleware(ThrottlingMiddleware(rate=1.0, on_throttle=on_throttle))

Works in both dp.run() (sync) and dp.run_async() (async) without any changes.

---

Async Mode

from gramix import Bot, Dispatcher, Router, load_env, F, ParseMode

load_env()
bot = Bot(parse_mode=ParseMode.HTML)
dp  = Dispatcher(bot)
rt  = Router()
dp.include(rt)

@rt.message("/start")
async def on_start(msg):
    msg.answer("Hello from async!")

@dp.on_startup
async def on_startup():
    print("Bot started.")

dp.run_async()

---

Webhook Mode

# Raw socket backend (no extra dependencies)
dp.run(webhook=True, webhook_url="https://yourdomain.com/", port=8080)

# With secret token validation
dp.run(webhook=True, webhook_url="https://yourdomain.com/", secret_token="mysecret", port=8080)

# aiohttp backend — pip install gramix[aiohttp]
dp.run(webhook=True, webhook_url="https://yourdomain.com/", backend="aiohttp", port=8080)

# FastAPI + uvicorn backend — pip install gramix[fastapi]
dp.run(webhook=True, webhook_url="https://yourdomain.com/", backend="fastapi", port=8080)

---

Lifecycle Hooks

@dp.on_startup
def on_startup():
    print("Bot is online.")

@dp.on_shutdown
def on_shutdown():
    print("Bot is shutting down.")

@dp.on_startup
async def async_startup():
    await db.connect()

---

API Reference

Full API reference is available on GitHub.

Bot — key methods

Method	Description
send_message(chat_id, text, *, keyboard, parse_mode, disable_preview, auto_split)	Send a text message.
send_photo(chat_id, photo, *, caption, keyboard, parse_mode)	Send a photo by file_id or URL.
send_video(chat_id, video, *, caption, keyboard, parse_mode)	Send a video.
send_game(chat_id, game_short_name, *, keyboard)	Send a Telegram game.
set_game_score(user_id, score, *, chat_id, message_id, force, disable_edit_return)	Set a user's score in a game.
get_game_high_scores(user_id, *, chat_id, message_id)	Get the high score table; returns list[GameHighScore].
edit_message_text(chat_id, message_id, text, *, keyboard, parse_mode)	Edit a message.
delete_message(chat_id, message_id)	Delete a message.
download_file(file_id)	Download file content as bytes.
send_poll(chat_id, question, options, *, poll_type, ...)	Send a regular poll or quiz.
send_location(chat_id, latitude, longitude, *, live_period, keyboard)	Send a location.
send_invoice(chat_id, title, description, payload, provider_token, currency, prices, ...)	Send a payment invoice.
set_webhook(url, *, secret_token)	Register a webhook URL.
close()	Close the HTTP client.

ThrottlingMiddleware

Parameter	Type	Default	Description
rate	float	1.0	Minimum seconds between accepted messages per user.
on_throttle	Callable | None	None	Optional callback (update) -> None (sync or async) invoked on throttle.

ParseMode

Constant	Value
ParseMode.HTML	"HTML"
ParseMode.MARKDOWN	"MarkdownV2"
ParseMode.MARKDOWN_LEGACY	"Markdown"

F — Filter Shortcuts

Filter	Matches when
F.text	Message has text
F.photo	Message has a photo
F.document	Message has a document
F.video	Message has a video
F.audio	Message has an audio file
F.voice	Message has a voice message
F.sticker	Message has a sticker
F.reply	Message is a reply
F.forward	Message is forwarded
F.private	Chat type is private
F.group	Chat type is group or supergroup
F.supergroup	Chat type is supergroup
F.channel	Chat type is channel
F.poll	Message contains a forwarded poll
F.location	Message contains a location
F.venue	Message contains a venue

---

License

MIT © riokzy