ΠΡΠΈΠ½Ρ ΡΠΎΠ½Π½ΡΠΉ Python-ΡΡΠ΅ΠΉΠΌΠ²ΠΎΡΠΊ Π΄Π»Ρ ΡΠ°Π·ΡΠ°Π±ΠΎΡΠΊΠΈ Π±ΠΎΡΠΎΠ² Π² MAX
ΠΠΎΠΊΡΠΌΠ΅Π½ΡΠ°ΡΠΈΡ
ΠΠ½ΡΠ΅ΡΡΠ΅ΠΉΡ ΠΎΡΠ½ΠΎΠ²Π°Π½ Π½Π° aiogram
maxo/dialogs ΡΠ΄Π΅Π»Π°Π½ΠΎ ΠΈΠ· aiogram_dialog
maxo/transport/webhook ΡΠ΄Π΅Π»Π°Π½ΠΎ ΠΈΠ· aiogram-webhook
Π§Π΅ΡΠ΅Π· pip:
pip install maxo
Π pyproject.toml:
[project]
dependencies = [
"maxo",
]- ΠΡΠΈΠ½Ρ
ΡΠΎΠ½Π½ΠΎΡΡΡ Π½Π° Π±Π°Π·Π΅
aiohttpΠΈunihttp(asyncio, PEP 492) - 100% ΠΏΠΎΠΊΡΡΡΠΈΠ΅ ΡΠΈΠΏΠ°ΠΌΠΈ,
adaptixΠ΄Π»Ρ Π²Π°Π»ΠΈΠ΄Π°ΡΠΈΠΈ Π΄Π°Π½Π½ΡΡ - Π ΠΎΡΡΠ΅ΡΡ, ΡΠΈΠ»ΡΡΡΡ, ΠΌΠΈΠ»Π΄Π²Π°ΡΠΈ
- ΠΡΡΡΠΎΠ΅Π½Π½Π°Ρ ΠΌΠ°ΡΠΈΠ½Π° ΡΠΎΡΡΠΎΡΠ½ΠΈΠΉ (FSM) ΠΈ Π΄ΠΈΠ°Π»ΠΎΠ³ΠΈ ΠΏΠΎΠ²Π΅ΡΡ Π½ΠΈΡ
- ΠΠΎΠ΄Π΄Π΅ΡΠΆΠΊΠ° Π»ΠΎΠ½Π³-ΠΏΠΎΠ»Π»ΠΈΠ½Π³Π° ΠΈ Π²Π΅Π±Ρ
ΡΠΊΠΎΠ² ΡΠ΅ΡΠ΅Π·
aiohttpΠΈfastapi - ΠΠ½ΡΠ΅Π³ΡΠ°ΡΠΈΠΈ Ρ
dishkaΠΈmagic_filter - ΠΠ²ΡΠΎΠ³Π΅Π½Π΅ΡΠ°ΡΠΈΡ ΠΌΠ΅ΡΠΎΠ΄ΠΎΠ², ΡΠΈΠΏΠΎΠ² ΠΈ Π°ΠΏΠ΄Π΅ΠΉΡΠΎΠ² ΠΏΠΎ ΠΎΡΠΈΡΠΈΠ°Π»ΡΠ½ΠΎΠΉ Π΄ΠΎΠΊΡΠΌΠ΅Π½ΡΠ°ΡΠΈΠΈ
ΠΠΎΠ»ΡΡΠ΅ ΠΏΡΠΈΠΌΠ΅ΡΠΎΠ² Π² ΠΏΡΠΈΠΌΠ΅ΡΠ°Ρ
from maxo import Bot, Dispatcher
from maxo.routing.updates import MessageCreated
from maxo.transport.long_polling import LongPolling
bot = Bot("TOKEN")
dp = Dispatcher()
@dp.message_created()
async def echo_handler(message: MessageCreated) -> None:
text = message.text or "Π’Π΅ΠΊΡΡΠ° Π½Π΅Ρ"
await message.answer(text)
LongPolling(dp).run(bot)from maxo import Bot, Dispatcher
from maxo.routing.filters import Command, DeeplinkFilter
from maxo.routing.updates import BotStarted, MessageCreated
from maxo.transport.long_polling import LongPolling
bot = Bot("TOKEN")
dp = Dispatcher()
@dp.bot_started(DeeplinkFilter())
async def deeplink_handler(bot_started: BotStarted, deeplink: str) -> None:
await bot_started.send_message(f"ΠΡΠΈΠ²Π΅Ρ! Π― Π±ΠΎΡ. ΠΠΈΠΏΠ»ΠΈΠ½ΠΊ: {deeplink}")
@dp.bot_started()
async def start_handler(bot_started: BotStarted) -> None:
await bot_started.send_message(f"ΠΡΠΈΠ²Π΅Ρ! Π― Π±ΠΎΡ. Π ΡΡ {bot_started.user.fullname}")
@dp.message(Command("help"))
async def help_handler(message: MessageCreated) -> None:
await message.send_message("ΠΠ° ΠΏΠΎΠΌΠΎΡΡΡ ΠΎΠ±ΡΠ°ΡΠ°ΠΉΡΠ΅ΡΡ Π² t.me/maxo_py")
LongPolling(dp).run(bot)from magic_filter import F
from maxo import Bot, Dispatcher
from maxo.integrations.magic_filter import MagicFilter
from maxo.routing.filters import CommandStart
from maxo.routing.updates import MessageCallback, MessageCreated
from maxo.transport.long_polling import LongPolling
from maxo.utils.builders import KeyboardBuilder
bot = Bot("TOKEN")
dp = Dispatcher()
@dp.message_created(CommandStart())
async def start_handler(message: MessageCreated) -> None:
maxo_url = "https://github.com/K1rL3s/maxo"
keyboard = (
KeyboardBuilder()
.add_callback(text="ΠΠΎΠ»Π±ΡΠΊ", payload="click_me")
.add_message(text="Π‘ΠΎΠΎΠ±ΡΠ΅Π½ΠΈΠ΅")
.add_link(text="ΠΠ΅ΡΠ΅ΠΉΡΠΈ Π² maxo", url=maxo_url)
.add_clipboard(text="Π‘ΠΊΠΎΠΏΠΈΡΠΎΠ²Π°ΡΡ maxo", payload=maxo_url)
.add_request_contact(text="ΠΠΎΠ΄Π΅Π»ΠΈΡΡΡΡ ΠΊΠΎΠ½ΡΠ°ΠΊΡΠ°ΠΌΠΈ")
.add_request_geo_location(text="ΠΠΎΠ΄Π΅Π»ΠΈΡΡΡΡ Π³Π΅ΠΎ ΠΏΠΎΠ·ΠΈΡΠΈΠ΅ΠΉ")
.adjust(2, 2, 1, 1)
)
await message.answer(text="ΠΠ½ΠΎΠΏΠΎΡΠΊΠΈ :3", keyboard=keyboard.build())
@dp.message_callback(MagicFilter(F.payload == "callback_payload"))
async def button_handler(callback: MessageCallback) -> None:
await callback.callback_answer("ΠΡ Π½Π°ΠΆΠ°Π»ΠΈ Π½Π° ΠΊΠ½ΠΎΠΏΠΊΡ!")
LongPolling(dp).run(bot)import logging
from aiohttp import web
from maxo import Bot, Dispatcher, Router
from maxo.enums import TextFormat
from maxo.routing.updates import BotStarted, MessageCreated
from maxo.routing.utils import collect_used_updates
from maxo.transport.webhook.adapters.aiohttp import AiohttpWebAdapter
from maxo.transport.webhook.engines import SimpleEngine, WebhookEngine
from maxo.transport.webhook.routing import StaticRouting
from maxo.transport.webhook.security import Security, StaticSecretToken
bot = Bot("TOKEN")
router = Router()
@router.bot_started()
async def start_handler(bot_started: BotStarted) -> None:
await bot_started.send_message(
text=f"ΠΡΠΈΠ²Π΅Ρ ΠΈΠ· Π²Π΅Π±Ρ
ΡΠΊΠ°, {bot_started.user.first_name}!",
)
@router.message_created()
async def echo_handler(message: MessageCreated) -> None:
await message.answer(
text=message.message.body.html_text,
format=TextFormat.HTML,
)
@router.after_startup()
async def on_startup(dispatcher: Dispatcher, webhook_engine: WebhookEngine) -> None:
await webhook_engine.set_webhook(update_types=collect_used_updates(dispatcher))
def main() -> None:
dispatcher = Dispatcher()
dispatcher.include(router)
engine = SimpleEngine(
dispatcher,
bot,
web_adapter=AiohttpWebAdapter(),
routing=StaticRouting(url="https://example.com/webhook"),
security=Security(secret_token=StaticSecretToken("pepa_pig")),
)
app = web.Application()
engine.register(app)
web.run_app(app, host="127.0.0.1", port=8080)
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
main()MAX - ΡΠΎΡΡΠΈΠΉΡΠΊΠΈΠΉ ΠΌΠ΅ΡΡΠ΅Π½Π΄ΠΆΠ΅Ρ. Π£ Π½Π΅Π³ΠΎ Π΅ΡΡΡ ΠΎΡΠΊΡΡΡΠΎΠ΅ Bot API, Π΄Π»Ρ ΡΠ°Π±ΠΎΡΡ Ρ ΠΊΠΎΡΠΎΡΡΠΌ ΠΈ ΡΠΎΠ·Π΄Π°Π½ maxo.
maxo - ΠΎΡΠ΄Π΅Π»ΡΠ½ΡΠΉ ΡΡΠ΅ΠΉΠΌΠ²ΠΎΡΠΊ ΠΈΠΌΠ΅Π½Π½ΠΎ Π΄Π»Ρ Π±ΠΎΡΠΎΠ² MAX, Π½ΠΎ ΠΈΠ½ΡΠ΅ΡΡΠ΅ΠΉΡ Π½Π°ΠΌΠ΅ΡΠ΅Π½Π½ΠΎ Π±Π»ΠΈΠ·ΠΎΠΊ ΠΊ aiogram, ΡΡΠΎΠ±Ρ ΠΏΠ΅ΡΠ΅Ρ
ΠΎΠ΄ Π±ΡΠ» ΠΌΠ°ΠΊΡΠΈΠΌΠ°Π»ΡΠ½ΠΎ Π±Π΅Π·Π±ΠΎΠ»Π΅Π·Π½Π΅Π½Π½ΡΠΌ. ΠΠΈΠ°Π»ΠΎΠ³ΠΈ (maxo.dialogs) ΠΏΠΎΡΡΠΈΡΠΎΠ²Π°Π½Ρ ΠΈΠ· aiogram_dialog, Π²Π΅Π±Ρ
ΡΠΊΠΈ (maxo.transport.webhook) - ΠΈΠ· aiogram-webhook.
ΠΠ°. ΠΠΎΠ΄Π΄Π΅ΡΠΆΠΈΠ²Π°Π΅ΡΡΡ ΠΈ long-polling, ΠΈ webhook ΡΠ΅ΡΠ΅Π· aiohttp ΠΈΠ»ΠΈ fastapi - ΡΠΌ. ΠΏΡΠΈΠΌΠ΅ΡΡ Π²ΡΡΠ΅.
Python 3.12, 3.13 ΠΈΠ»ΠΈ 3.14.
ΠΠ° ΠΏΠ»Π°ΡΡΠΎΡΠΌΠ΅ Π΄Π»Ρ ΠΏΠ°ΡΡΠ½ΡΡΠΎΠ².
FSM Π²ΡΡΡΠΎΠ΅Π½Π° Π² maxo - Π΅ΡΡΡ MemoryStorage ΠΈΠ· ΠΊΠΎΡΠΎΠ±ΠΊΠΈ ΠΈ ΠΎΠΏΡΠΈΠΎΠ½Π°Π»ΡΠ½ΠΎΠ΅ Ρ
ΡΠ°Π½ΠΈΠ»ΠΈΡΠ΅ Π² Redis (maxo[redis]). ΠΠΎΠ΄ΡΠΎΠ±Π½ΠΎΡΡΠΈ - Π² Π΄ΠΎΠΊΡΠΌΠ΅Π½ΡΠ°ΡΠΈΠΈ.
ΠΡΠ»ΠΈ Ρ Π²Π°Ρ Π΅ΡΡΡ Π²ΠΎΠΏΡΠΎΡΡ, Π²Ρ ΠΌΠΎΠΆΠ΅ΡΠ΅ Π·Π°Π΄Π°ΡΡ ΠΈΡ Π² Π’Π΅Π»Π΅Π³ΡΠ°ΠΌΠ΅ @maxo_py ΠΈΠ»ΠΈ ΠΠ°ΠΊΡΠ΅