ΠΡΠΈΠ½Ρ ΡΠΎΠ½Π½ΡΠΉ 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
from maxo.utils.facades import MessageCreatedFacade
bot = Bot("TOKEN")
dispatcher = Dispatcher()
@dispatcher.message_created()
async def echo_handler(update: MessageCreated, facade: MessageCreatedFacade) -> None:
text = update.message.body.text or "Π’Π΅ΠΊΡΡΠ° Π½Π΅Ρ"
await facade.answer_text(text)
LongPolling(dispatcher).run(bot)from maxo import Bot, Dispatcher
from maxo.routing.filters import Command, CommandObject, CommandStart
from maxo.routing.updates import MessageCreated
from maxo.transport.long_polling import LongPolling
from maxo.utils.facades import MessageCreatedFacade
bot = Bot("TOKEN")
dispatcher = Dispatcher()
@dispatcher.message_created(CommandStart())
# ΠΈΠ»ΠΈ @dispatcher.message_created(Command("start"))
async def start_handler(
message: MessageCreated,
command: CommandObject,
facade: MessageCreatedFacade,
) -> None:
await facade.answer_text(f"ΠΡΠΈΠ²Π΅Ρ! Π― Π±ΠΎΡ. ΠΠΈΠΏΠ»ΠΈΠ½ΠΊ: {command.args}")
LongPolling(dispatcher).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
from maxo.utils.facades import MessageCallbackFacade, MessageCreatedFacade
bot = Bot("TOKEN")
dispatcher = Dispatcher()
@dispatcher.message_created(CommandStart())
async def start_handler(message: MessageCreated, facade: MessageCreatedFacade) -> None:
keyboard = (
KeyboardBuilder()
.add_callback(
text="ΠΠΎΠ»Π±ΡΠΊ ΠΊΠ½ΠΎΠΏΠΊΠ°",
payload="callback_payload",
)
.add_message(text="Π’Π΅ΠΊΡΡΠΎΠ²Π°Ρ ΠΊΠ½ΠΎΠΏΠΊΠ°")
.add_link(
text="Maxo",
url="https://github.com/K1rL3s/maxo",
)
.adjust(1, repeat=True)
.build()
)
await facade.answer_text("ΠΠ½ΠΎΠΏΠΎΡΠΊΠΈ:", keyboard=keyboard)
@dispatcher.message_callback(MagicFilter(F.payload == "callback_payload"))
async def button_handler(
callback: MessageCallback,
facade: MessageCallbackFacade,
) -> None:
await facade.callback_answer("ΠΡ Π½Π°ΠΆΠ°Π»ΠΈ Π½Π° ΠΊΠ½ΠΎΠΏΠΊΡ!")
LongPolling(dispatcher).run(bot)import logging
import os
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
from maxo.utils.facades import BotStartedFacade, MessageCreatedFacade
bot = Bot(os.environ["TOKEN"])
router = Router()
@router.bot_started()
async def start_handler(bot_started: BotStarted, facade: BotStartedFacade) -> None:
await facade.send_message(text=f"ΠΡΠΈΠ²Π΅Ρ ΠΈΠ· Π²Π΅Π±Ρ
ΡΠΊΠ°, {bot_started.user.first_name}!")
@router.message_created()
async def echo_handler(message: MessageCreated, facade: MessageCreatedFacade) -> None:
await facade.answer_text(
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 ΠΈΠ»ΠΈ ΠΠ°ΠΊΡΠ΅