API reference

Bot and application

Platform-neutral Bot facade for Phase 2.

class peyk.bot.base.ClientLifecycle(*args, **kwargs)

Bases: Protocol

ClientLifecycle provides the bot API surface used by peyk.

async close() → None

Performs the close operation for the bot client.

class peyk.bot.base.BotDefaults(parse_mode: ParseMode | str | None = None, link_preview: bool | None = None, protect_content: bool | None = None, style_fallback: StyleFallback = StyleFallback.NONE)

Bases: object

Defaults applied only when the active capability registry confirms them.

parse_mode: ParseMode | str | None = None

link_preview: bool | None = None

protect_content: bool | None = None

style_fallback: StyleFallback = 'none'

class peyk.bot.base.Bot(token: str, *, platform: Literal['bale', 'telegram', 'rubika'], defaults: BotDefaults | None = None, on_unsupported: UnsupportedPolicy = UnsupportedPolicy.DEFAULT, session: Session | None = None, retry_policy: RetryPolicy | None = None, logger: Logger | None = None, base_url: str | None = None)

Bases: Generic[ClientT]

Unified async bot facade over one audited platform client.

Example

bot = Bot("TOKEN", platform="telegram")

@bot.command("start")
async def start(message):
    await bot.send_message(message.chat.id, "Hello")

bot.run()

property client: ClientT

Return the lazily-created raw platform client.

supports(feature: Feature) → bool

Return whether the audited capability is usable.

async close() → None

Close the underlying client/session when it has been initialized.

async set_webhook(url: str, *, max_connections: int | None = None, allowed_updates: Sequence[str] | None = None, drop_pending_updates: bool | None = None, secret_token: str | None = None) → bool

Set the platform webhook after checking its audited capabilities.

secret_token is enforceable by Telegram only. Bale and Rubika use the framework-level path secret instead and therefore reject this platform-specific option rather than silently ignoring it.

async delete_webhook(*, drop_pending_updates: bool | None = None) → bool

Remove the configured webhook when the platform exposes that operation.

async get_webhook_info() → WebhookInfo | WebhookInfo

Return the native webhook-info model where the platform exposes it.

async me() → User

Fetch and cache the neutral authenticated bot identity.

property id: int | str

Return the authenticated bot ID after me() has been fetched.

normalize_update(raw: Update | Update | Update | InlineMessage) → Message | CallbackQuery | IncomingChatMemberStatusUpdate | IncomingPreCheckoutQuery | IncomingShippingQuery | IncomingMessageDeleted | IncomingBotMembershipChange | Update | Update | Update

Normalize a raw platform update and bind supported events to this bot.

command(*names: str, prefix: str = '/') → Callable[[Callable[[HandlerP], HandlerR]], Callable[[HandlerP], HandlerR]]

Register one handler for one or more command names.

Each alias is registered independently so all aliases share the exact original Python callable and its type signature. This delegates to peyk.dispatcher.router.Router.command() on the bot’s router.

message(*filters: object) → Callable[[Callable[[HandlerP], HandlerR]], Callable[[HandlerP], HandlerR]]

Register a message handler on the bot’s plain router.

callback_query(*filters: object) → Callable[[Callable[[HandlerP], HandlerR]], Callable[[HandlerP], HandlerR]]

Register a callback-query handler on the bot’s plain router.

callback(data: str, *, prefix: bool = False) → Callable[[Callable[[HandlerP], HandlerR]], Callable[[HandlerP], HandlerR]]

Register a callback handler for exact data or a data prefix.

on_startup(callback: Callable[[], Awaitable[None] | None]) → Callable[[], Awaitable[None] | None]

Register a startup callback using the bare-decorator form.

on_shutdown(callback: Callable[[], Awaitable[None] | None]) → Callable[[], Awaitable[None] | None]

Register a shutdown callback using the bare-decorator form.

include_router(router: Router) → Router

Include another router in the bot’s router tree.

async run_async(**polling_options: object) → None

Validate the token, log startup, and run automatic polling asynchronously.

run(**polling_options: object) → None

Run this bot’s router with automatic polling in a blocking call.

async send_message(chat_id: int | str, text: str | 'RichText' | Text, *, reply_to_message_id: int | str | None = None, reply_markup: object | None = None, parse_mode: ParseMode | str | None = None, **kwargs: object) → Message

Send plain or composable text and resolve formatting for the active platform.

async send_photo(chat_id: int | str, photo: object, *, caption: str | 'RichText' | Text | None = None, parse_mode: ParseMode | str | None = None, **kwargs: object) → Message

Sends photo through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• photo – Photo input supplied to the operation.

• caption – Value used by this operation.

• parse_mode – Value used by this operation.

Returns:

Result produced by the bot operation.

async send_video(chat_id: int | str, video: object, *, caption: str | 'RichText' | Text | None = None, parse_mode: ParseMode | str | None = None, **kwargs: object) → Message

Sends video through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• video – Video input supplied to the operation.

• caption – Value used by this operation.

• parse_mode – Value used by this operation.

Returns:

Result produced by the bot operation.

async send_audio(chat_id: int | str, audio: object, *, caption: str | 'RichText' | Text | None = None, parse_mode: ParseMode | str | None = None, **kwargs: object) → Message

Sends audio through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• audio – Audio input supplied to the operation.

• caption – Value used by this operation.

• parse_mode – Value used by this operation.

Returns:

Result produced by the bot operation.

async send_voice(chat_id: int | str, voice: object, *, caption: str | 'RichText' | Text | None = None, parse_mode: ParseMode | str | None = None, **kwargs: object) → Message

Sends voice through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• voice – Voice input supplied to the operation.

• caption – Value used by this operation.

• parse_mode – Value used by this operation.

Returns:

Result produced by the bot operation.

async send_document(chat_id: int | str, document: object, *, caption: str | 'RichText' | Text | None = None, parse_mode: ParseMode | str | None = None, **kwargs: object) → Message

Sends document through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• document – Document input supplied to the operation.

• caption – Value used by this operation.

• parse_mode – Value used by this operation.

Returns:

Result produced by the bot operation.

async edit_message_text(chat_id: int | str, message_id: int | str, text: str | 'RichText' | Text, *, parse_mode: ParseMode | str | None = None, **kwargs: object) → Message

Edits message text through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• message_id – Identifier of the target message.

• text – Text content supplied to the operation.

• parse_mode – Value used by this operation.

Returns:

Result produced by the bot operation.

async send_contact(chat_id: int | str, phone_number: str, first_name: str, **kwargs: object) → Message

Sends contact through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• phone_number – Value used by this operation.

• first_name – Value used by this operation.

Returns:

Result produced by the bot operation.

async send_location(chat_id: int | str, latitude: float, longitude: float, **kwargs: object) → Message

Sends location through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• latitude – Value used by this operation.

• longitude – Value used by this operation.

Returns:

Result produced by the bot operation.

async send_poll(chat_id: int | str, question: str, options: Sequence[str], **kwargs: object) → Message

Sends poll through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• question – Value used by this operation.

• options – Value used by this operation.

Returns:

Result produced by the bot operation.

async edit_message_reply_markup(chat_id: int | str, message_id: int | str, **kwargs: object) → Message

Edits message reply markup through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• message_id – Identifier of the target message.

Returns:

Result produced by the bot operation.

async delete_message(chat_id: int | str, message_id: int | str) → bool

Removes message through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• message_id – Identifier of the target message.

Returns:

Result produced by the bot operation.

async forward_message(chat_id: int | str, from_chat_id: int | str, message_id: int | str) → Message

Performs the forward message operation for the bot client.

Parameters:

• chat_id – Identifier of the target chat.

• from_chat_id – Value used by this operation.

• message_id – Identifier of the target message.

Returns:

Result produced by the bot operation.

async send_media_group(chat_id: int | str, media: Sequence[object], **kwargs: object) → list[Message]

Send a media group when the active platform advertises album support.

async answer_callback_query(callback_query_id: str, *, text: str | None = None, show_alert: bool = False) → bool

Answers the callback query request through the bot API.

Parameters:

• callback_query_id – Identifier of the callback query.

• text – Text content supplied to the operation.

• show_alert – Value used by this operation.

Returns:

Result produced by the bot operation.

async send_chat_action(chat_id: int | str, action: str) → bool

Sends chat action through the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• action – Value used by this operation.

Returns:

Result produced by the bot operation.

async get_chat(chat_id: int | str) → Chat

Retrieves chat from the bot API.

Parameters:

chat_id – Identifier of the target chat.

Returns:

Result produced by the bot operation.

async get_chat_member(chat_id: int | str, user_id: int) → ChatMember

Retrieves chat member from the bot API.

Parameters:

• chat_id – Identifier of the target chat.

• user_id – Identifier of the target user.

Returns:

Result produced by the bot operation.

async ban_chat_member(chat_id: int | str, user_id: int) → bool

Performs the ban chat member operation for the bot client.

Parameters:

• chat_id – Identifier of the target chat.

• user_id – Identifier of the target user.

Returns:

Result produced by the bot operation.

async unban_chat_member(chat_id: int | str, user_id: int) → bool

Performs the unban chat member operation for the bot client.

Parameters:

• chat_id – Identifier of the target chat.

• user_id – Identifier of the target user.

Returns:

Result produced by the bot operation.

async get_file(file_id: str) → File

Return a neutral file descriptor.

async download(file_id: str) → bytes

Download file bytes using the platform-specific file descriptor/URL.

Dispatcher and Router

class peyk.dispatcher.dispatcher.Dispatcher(*, name: str | None = None, storage: BaseStorage | None = None, fsm_strategy: FSMStrategy | str = FSMStrategy.USER_IN_CHAT, events_isolation: BaseEventIsolation | None = None, **workflow_data: object)

Bases: Router

Root router that feeds normalized updates from one or more bots.

The dispatcher owns polling tasks but deliberately delegates event matching to Router; Phase 3 does not alter router matching semantics.

resolve_used_update_types() → list[str]

Return the Telegram update types this dispatcher should receive.

Telegram remembers the allowed_updates last sent for a bot token, whether it came from getUpdates, from setWebhook or from a completely different program that used the same token earlier. If a new run does not send the parameter, the old filter silently stays in effect, so e.g. callback_query and inline_query updates never arrive even though the handlers are registered. Peyk therefore always sends an explicit list.

The list contains every update type Peyk can parse, except the three opt-in types Telegram does not deliver by default (chat_member, message_reaction and message_reaction_count); those are added only when a router in the tree has a handler that needs them.

Returns:

Update type names in Telegram’s allowed_updates spelling.

async feed_raw_update(bot: Bot[object], raw_update: object) → object

Normalize one raw update, inject dispatcher context, and propagate it.

async start_polling(*bots: Bot[object], polling_timeout: int = 30, allowed_updates: Sequence[str] | None = None, skip_updates: bool = False, handle_signals: bool = True, close_bots: bool = True, handle_as_tasks: bool = False, max_concurrent_updates: int = 100) → None

Poll all supplied bots concurrently until stopped or cancelled.

run_polling(*bots: Bot[object], **kwargs: object) → None

Run start_polling() through asyncio.run.

async start_webhook(*bots: Bot[object], base_url: str, host: str = '0.0.0.0', port: int = 8080, path_prefix: str = '/webhook', secrets: Sequence[str] | None = None, ssl_context: object | None = None, allowed_updates: Sequence[str] | None = None, drop_pending_updates: bool = False, manage_registration: bool = True, delete_on_shutdown: bool = False, max_connections: int | None = None, handle_in_background: bool = True) → None

Serve Telegram, Bale and Rubika webhooks from one aiohttp app.

Each bot receives a non-token URL of {path_prefix}/{platform}/{secret}. Rubika’s confirmed inline endpoint is registered at the additional /inline path.

run_webhook(*bots: Bot[object], **kwargs: object) → None

Run start_webhook() through asyncio.run for Windows-safe automation.

async stop_polling() → None

Request graceful polling shutdown after the current batch.

class peyk.dispatcher.router.Observer(router: Router, name: str)

Bases: object

Handler collection for one event type, with filters and middleware.

register(handler: Callable[[P], R], *filters: object) → Callable[[P], R]

Performs the register operation for the dispatcher client.

Parameters:

handler – Value used by this operation.

Returns:

Result produced by the dispatcher operation.

filter(*filters: object) → Observer

Performs the filter operation for the dispatcher client.

Returns:

Result produced by the dispatcher operation.

middleware(middleware: BaseMiddleware) → BaseMiddleware

Performs the middleware operation for the dispatcher client.

Parameters:

middleware – Value used by this operation.

Returns:

Result produced by the dispatcher operation.

outer_middleware(middleware: BaseMiddleware) → BaseMiddleware

Performs the outer middleware operation for the dispatcher client.

Parameters:

middleware – Value used by this operation.

Returns:

Result produced by the dispatcher operation.

class peyk.dispatcher.router.Router(*, name: str | None = None)

Bases: object

Aiogram-style first-match router with nested ordered subrouters.

command(*names: str, prefix: str = '/') → Callable[[Callable[[P], R]], Callable[[P], R]]

Register a message handler for one or more bot commands.

This is shorthand for router.message(Command(name, prefix=prefix)) and is available on Router, Dispatcher and (through Bot.command()) on the bot’s own router. Each alias is registered independently, so all of them share the same Python callable:

@router.command("start", "help")
async def start(message): ...

include_router(router: Router) → Router

Performs the include router operation for the dispatcher client.

Parameters:

router – Value used by this operation.

Returns:

Result produced by the dispatcher operation.

middleware(middleware: BaseMiddleware) → BaseMiddleware

Register outer whole-subtree middleware, preserving the D3 API.

handler_middleware(middleware: BaseMiddleware) → BaseMiddleware

Register middleware that runs after handler flags have been injected.

has_handlers() → bool

Performs the has handlers operation for the dispatcher client.

Returns:

Result produced by the dispatcher operation.

async propagate_platform_event(update: object, *, platform: str, **data: object) → object

Dispatch a native platform update and its named Telegram field observer.

platform_event receives the complete native update object. For Telegram, a named observer receives the native model stored in the corresponding Update field; other platforms currently have no named-native observer surface.

async propagate_event(event: object, **data: object) → object

Dispatch one event using first-match-wins semantics.

Types

Neutral inbound bot objects and their bound actions.

class peyk.types.core.User(id: int | str, is_bot: bool, first_name: str, last_name: str | None = None, username: str | None = None, language_code: str | None = None, raw: object = None)

Bases: object

Platform-neutral user identity.

id: int | str

is_bot: bool

first_name: str

last_name: str | None = None

username: str | None = None

language_code: str | None = None

raw: object = None

class peyk.types.core.Chat(id: int | str, type: ChatType, title: str | None = None, username: str | None = None, language_code: str | None = None, raw: object = None)

Bases: object

Platform-neutral chat identity and display metadata.

id: int | str

type: ChatType

title: str | None = None

username: str | None = None

language_code: str | None = None

raw: object = None

class peyk.types.core.File(id: str, path: str | None = None, name: str | None = None, size: int | None = None, raw: object = None)

Bases: object

Platform-neutral file descriptor returned by Bot.get_file.

id: str

path: str | None = None

name: str | None = None

size: int | None = None

raw: object = None

class peyk.types.core.ChatMember(status: str, user: User | None = None, raw: object = None)

Bases: object

Platform-neutral chat membership status.

status: str

user: User | None = None

raw: object = None

class peyk.types.core.Message(message_id: Identifier | None = None, chat_id: Identifier | None = None, chat_type: str | None = None, sender_id: Identifier | None = None, text: str | None = None, date: int | None = None, is_edited: bool = False, update_kind: str = 'message', reply_to_message_id: Identifier | None = None, media: object = None, new_chat_members: list[Identifier] | None = None, left_chat_member: Identifier | None = None, successful_payment: object = None, raw: object = None, from_user: User | None = None, chat: Chat | None = None, content_type: ContentType = ContentType.UNKNOWN, bot: 'Bot' | None = None)

Bases: IncomingMessage, _BoundActions

Normalized message with a bound peyk.Bot action surface.

from_user: User | None = None

chat: Chat | None = None

content_type: ContentType = 'unknown'

bot: 'Bot' | None = None

async answer(text: TextContent, **kwargs: object) → Message

Send a text message to this message’s chat.

async reply(text: TextContent, **kwargs: object) → Message

Alias for answer().

async edit_text(text: TextContent, **kwargs: object) → Message

Edit this message’s text.

async delete() → bool

Delete this message.

async forward(chat_id: int | str) → Message

Forward this message to chat_id.

async answer_photo(photo: object, *, caption: TextContent | None = None, **kwargs: object) → Message

Send a photo in this message’s chat.

async answer_video(video: object, *, caption: TextContent | None = None, **kwargs: object) → Message

Send a video in this message’s chat.

async answer_audio(audio: object, *, caption: TextContent | None = None, **kwargs: object) → Message

Send audio in this message’s chat.

async answer_voice(voice: object, *, caption: TextContent | None = None, **kwargs: object) → Message

Send a voice message in this message’s chat.

async answer_document(document: object, *, caption: TextContent | None = None, **kwargs: object) → Message

Send a document in this message’s chat.

async answer_chat_action(action: str) → bool

Send a chat action to this message’s chat.

class peyk.types.core.CallbackQuery(id: str | None = None, from_user_id: Identifier | None = None, chat_id: Identifier | None = None, message_id: Identifier | None = None, inline_message_id: str | None = None, data: str | None = None, raw: object = None, from_user: User | None = None, message: Message | None = None, bot: 'Bot' | None = None)

Bases: IncomingCallbackQuery, _BoundActions

Normalized callback event with message actions when a message exists.

from_user: User | None = None

message: Message | None = None

bot: 'Bot' | None = None

async answer(text: str | None = None, show_alert: bool = False) → bool

Answer this callback, degrading cosmetic toast/alert differences by policy.

Filters

Aiogram-style platform-neutral filters.

class peyk.filters.BaseFilter

Bases: ABC

Base class for filters that may also contribute handler data.

peyk.filters.and_f(*filters: object) → BaseFilter

Provides the and f operation for the peyk integration.

Returns:

Result produced by the operation.

peyk.filters.or_f(*filters: object) → BaseFilter

Provides the or f operation for the peyk integration.

Returns:

Result produced by the operation.

peyk.filters.invert_f(filter_: object) → BaseFilter

Provides the invert f operation for the peyk integration.

Parameters:

filter – Value used by this operation.

Returns:

Result produced by the operation.

peyk.filters.normalize_filter(value: object) → BaseFilter

Provides the normalize filter operation for the peyk integration.

Parameters:

value – Value used by this operation.

Returns:

Result produced by the operation.

class peyk.filters.Command(*values: str, prefix: str = '/', ignore_case: bool = False, ignore_mention: bool = False, magic: object | None = None)

Bases: BaseFilter

Match a command and expose its parsed CommandObject as command.

class peyk.filters.CommandStart(deep_link: bool = False, **kwargs: object)

Bases: Command

Match the /start command, optionally requiring deep-link arguments.

class peyk.filters.CommandObject(prefix: str, command: str, mention: str | None = None, args: str | None = None, regexp_match: Match[str] | None = None, magic_result: object | None = None)

Bases: object

Parsed command information supplied to a matching handler.

args: str | None = None

magic_result: object | None = None

mention: str | None = None

regexp_match: Match[str] | None = None

prefix: str

command: str

class peyk.filters.CallbackData

Bases: BaseFilter

Dataclass-based callback-data factory compatible with aiogram’s model.

classmethod filter(magic: object | None = None) → BaseFilter

Provides the filter operation for the peyk integration.

Parameters:

magic – Value used by this operation.

Returns:

Result produced by the operation.

pack() → str

Provides the pack operation for the peyk integration.

Returns:

Result produced by the operation.

classmethod unpack(value: str) → CallbackDataT

Provides the unpack operation for the peyk integration.

Parameters:

value – Value used by this operation.

Returns:

Result produced by the operation.

class peyk.filters.CallbackDataEquals(value: str)

Bases: BaseFilter

Match an exact callback-data string.

class peyk.filters.CallbackDataStartsWith(prefix: str)

Bases: BaseFilter

Match callback data beginning with a prefix.

class peyk.filters.ChatTypeFilter(*chat_types: ChatType | str)

Bases: BaseFilter

Match a neutral chat type, treating Telegram supergroup as group.

peyk.filters.ChatType

alias of ChatTypeFilter

class peyk.filters.StateFilter(*states: State | type[StatesGroup] | str | None | Pattern[str], storage: BaseStorage | None = None, platform: str | None = None)

Bases: BaseFilter

Match raw_state injected by FSMContextMiddleware.

StateGroup values match any state in the group, re.Pattern values use match(), None matches the default empty state, and peyk.fsm.state.any_state matches every state.

class peyk.filters.MagicData(magic: object)

Bases: BaseFilter

Evaluate a magic filter against dispatcher dependency data.

class peyk.filters.ExceptionTypeFilter(*types: type[BaseException])

Bases: BaseFilter

Match an ErrorEvent by exception type.

class peyk.filters.ExceptionMessageFilter(message: str, *, contains: bool = False)

Bases: BaseFilter

Match an ErrorEvent by exception message.

class peyk.filters.PlatformFilter(*platforms: str)

Bases: BaseFilter

Match events handled by one of the named platforms.

class peyk.filters.SupportsFilter(feature: Feature)

Bases: BaseFilter

Match when the injected bot advertises audited support for a feature.

class peyk.filters.TextEquals(value: str, case_sensitive: bool = True)

Bases: BaseFilter

Match a message whose text equals value.

class peyk.filters.TextContains(value: str, case_sensitive: bool = True)

Bases: BaseFilter

Match a message containing value.

Keyboard

class peyk.keyboard.builder.InlineKeyboardBuilder

Bases: _Builder[InlineButton]

Build an inline keyboard without selecting a platform.

button(**fields: object) → InlineKeyboardBuilder

Performs the button operation for the keyboard client.

Returns:

Result produced by the keyboard operation.

as_markup(**options: object) → InlineKeyboard

Performs the as markup operation for the keyboard client.

Returns:

Result produced by the keyboard operation.

classmethod from_markup(markup: InlineKeyboard) → InlineKeyboardBuilder

Performs the from markup operation for the keyboard client.

Parameters:

markup – Value used by this operation.

Returns:

Result produced by the keyboard operation.

class peyk.keyboard.builder.ReplyKeyboardBuilder

Bases: _Builder[ReplyButton]

Build a reply keyboard without selecting a platform.

button(**fields: object) → ReplyKeyboardBuilder

Performs the button operation for the keyboard client.

Returns:

Result produced by the keyboard operation.

as_markup(**options: object) → ReplyKeyboard

Performs the as markup operation for the keyboard client.

Returns:

Result produced by the keyboard operation.

classmethod from_markup(markup: ReplyKeyboard) → ReplyKeyboardBuilder

Performs the from markup operation for the keyboard client.

Parameters:

markup – Value used by this operation.

Returns:

Result produced by the keyboard operation.

Formatting

aiogram-style composable formatting built on Peyk’s RichText IR.

The composition objects are platform-neutral. Rendering happens only when a target platform is known, so application code does not carry platform names.

class peyk.formatting.Text(*body: str | RichText | Text, sep: str = '')

Bases: object

A composable sequence of text fragments and formatting nodes.

body: tuple[str | RichText | Text, ...]

to_rich_text() → RichText

Compile this composition into Peyk’s platform-neutral IR.

line() → Text

Append a newline and return a new composition.

bold(*body: str | RichText | Text) → Text

Append a bold fragment and return a new composition.

italic(*body: str | RichText | Text) → Text

Append an italic fragment and return a new composition.

underline(*body: str | RichText | Text) → Text

Append an underlined fragment and return a new composition.

strikethrough(*body: str | RichText | Text) → Text

Append a strikethrough fragment and return a new composition.

spoiler(*body: str | RichText | Text) → Text

Append a spoiler fragment and return a new composition.

code(*body: str | RichText | Text) → Text

Append an inline-code fragment and return a new composition.

pre(*body: str | RichText | Text, language: str | None = None) → Text

Append a preformatted fragment and return a new composition.

link(text: str | RichText | Text, url: str) → Text

Append a URL link and return a new composition.

mention(text: str | RichText | Text, user_id: int | str) → Text

Append a Telegram user mention and return a new composition.

blockquote(*body: str | RichText | Text) → Text

Append a block quote and return a new composition.

expandable_blockquote(*body: str | RichText | Text) → Text

Append an expandable block quote and return a new composition.

render(target: PlatformCapabilities | str) → str

Render this composition for a platform or audited capability object.

as_kwargs(target: PlatformCapabilities | str) → dict[str, object]

Return the exact text-bearing keyword arguments for target.

class peyk.formatting.Bold(*body: str | RichText | Text, sep: str = '')

Bases: Text

Render children in bold.

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.Italic(*body: str | RichText | Text, sep: str = '')

Bases: Text

Render children in italics.

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.Underline(*body: str | RichText | Text, sep: str = '')

Bases: Text

Render children with an underline where the target supports it.

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.Strikethrough(*body: str | RichText | Text, sep: str = '')

Bases: Text

Render children with a strikethrough where the target supports it.

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.Spoiler(*body: str | RichText | Text, sep: str = '')

Bases: Text

Render children as a spoiler where the target supports it.

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.Code(*body: str | RichText | Text, sep: str = '')

Bases: Text

Render children as inline code.

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.Pre(*body: str | RichText | Text, language: str | None = None)

Bases: Text

Render children as a preformatted block.

language: str | None

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.TextLink(*body: str | RichText | Text, url: str)

Bases: Text

Render children as a URL link.

url: str

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.TextMention(*body: str | RichText | Text, user_id: int | str)

Bases: Text

Render children as a Telegram user mention.

user_id: int | str

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.BlockQuote(*body: str | RichText | Text, sep: str = '')

Bases: Text

Render children as a block quote.

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

class peyk.formatting.ExpandableBlockQuote(*body: str | RichText | Text, sep: str = '')

Bases: Text

Render children as an expandable block quote.

to_rich_text() → RichText

Provides the to rich text operation for the peyk integration.

Returns:

Result produced by the operation.

peyk.formatting.as_line(*items: str | RichText | Text, sep: str = ' ') → Text

Join items on one line.

peyk.formatting.as_list(*items: str | RichText | Text | Iterable[str | RichText | Text], sep: str = '\n') → Text

Join items as a list separated by sep; one iterable is also accepted.

peyk.formatting.as_marked_list(*items: str | ~peyk.utils.text_formatting.RichText | ~peyk.formatting.Text | ~typing.Iterable[str | ~peyk.utils.text_formatting.RichText | ~peyk.formatting.Text], marker: str = '▫️', se, sep: str = ) ->) → Text

Prefix each item with marker.

peyk.formatting.as_numbered_list(*items: str | RichText | Text | Iterable[str | RichText | Text], sep: str = '\n') → Text

Prefix each item with its one-based position.

peyk.formatting.as_section(title: str | RichText | Text, *body: str | RichText | Text, sep: str = '\n') → Text

Return a title followed by body lines.

peyk.formatting.as_marked_section(title: str | ~peyk.utils.text_formatting.RichText | ~peyk.formatting.Text, *body: str | ~peyk.utils.text_formatting.RichText | ~peyk.formatting.Text, marker: str = '▫️', se, sep: str = ) ->) → Text

Return a title followed by a marked list.

peyk.formatting.as_key_value(key: str | RichText | Text, value: str | RichText | Text, sep: str = ': ') → Text

Join a key and value with sep.

peyk.formatting.resolve_text(content: str | RichText | Text, platform_caps: PlatformCapabilities, policy: UnsupportedPolicy, defaults: TextDefaults) → tuple[str, dict[str, object]]

Resolve text once at send time into target text and native keyword arguments.

Plain strings remain plain unless a default parse mode is configured. Rich compositions are rendered from the neutral IR; Rubika additionally receives its structured metadata. Explicit parse-mode handling is applied by peyk.Bot before this resolver is called.

class peyk.formatting.TextDefaults(*args, **kwargs)

Bases: Protocol

Minimal defaults contract required by resolve_text().

parse_mode: ParseMode | str | None

peyk.formatting.bold(text: str | RichText) → RichText

Performs the bold operation for the utility client.

Parameters:

text – Text content supplied to the operation.

Returns:

Result produced by the utility operation.

peyk.formatting.italic(text: str | RichText) → RichText

Performs the italic operation for the utility client.

Parameters:

text – Text content supplied to the operation.

Returns:

Result produced by the utility operation.

peyk.formatting.underline(text: str | RichText) → RichText

Performs the underline operation for the utility client.

Parameters:

text – Text content supplied to the operation.

Returns:

Result produced by the utility operation.

peyk.formatting.strikethrough(text: str | RichText) → RichText

Performs the strikethrough operation for the utility client.

Parameters:

text – Text content supplied to the operation.

Returns:

Result produced by the utility operation.

peyk.formatting.spoiler(text: str | RichText) → RichText

Performs the spoiler operation for the utility client.

Parameters:

text – Text content supplied to the operation.

Returns:

Result produced by the utility operation.

peyk.formatting.code(text: str | RichText) → RichText

Performs the code operation for the utility client.

Parameters:

text – Text content supplied to the operation.

Returns:

Result produced by the utility operation.

peyk.formatting.pre(text: str | RichText, language: str | None = None) → RichText

Performs the pre operation for the utility client.

Parameters:

• text – Text content supplied to the operation.

• language – Value used by this operation.

Returns:

Result produced by the utility operation.

peyk.formatting.link(text: str | RichText, url: str) → RichText

Performs the link operation for the utility client.

Parameters:

• text – Text content supplied to the operation.

• url – Target URL.

Returns:

Result produced by the utility operation.

peyk.formatting.mention_user(text: str | RichText, user_id: int | str) → RichText

Performs the mention user operation for the utility client.

Parameters:

• text – Text content supplied to the operation.

• user_id – Identifier of the target user.

Returns:

Result produced by the utility operation.

peyk.formatting.mention_username(text: str | RichText, username: str) → RichText

Performs the mention username operation for the utility client.

Parameters:

• text – Text content supplied to the operation.

• username – Value used by this operation.

Returns:

Result produced by the utility operation.

peyk.formatting.bale_expandable(title: str, text: str | RichText) → RichText

Performs the bale expandable operation for the utility client.

Parameters:

• title – Title to apply to the target resource.

• text – Text content supplied to the operation.

Returns:

Result produced by the utility operation.

class peyk.formatting.ParseMode(*values)

Bases: str, Enum

ParseMode defines a public API type for peyk.

MARKDOWN = 'Markdown'

MARKDOWN_V2 = 'MarkdownV2'

HTML = 'HTML'

Webhook

class peyk.webhook.IPFilter(networks: Sequence[str], *, trust_forwarded_for: bool = False)

Bases: object

Allow webhook requests only when their source IP is in configured networks.

trust_forwarded_for is deliberately opt-in. When disabled, only the transport peer address is considered, so an untrusted client cannot choose its own address through X-Forwarded-For.

allows(peer_ip: str, headers: Mapping[str, str]) → bool

Return whether the request source belongs to an allowed network.

class peyk.webhook.WebhookOptions(allowed_updates: list[str] | None = None, drop_pending_updates: bool = False, max_connections: int | None = None)

Bases: object

Platform-neutral webhook registration options.

allowed_updates: list[str] | None = None

drop_pending_updates: bool = False

max_connections: int | None = None

class peyk.webhook.WebhookProcessor(dispatcher: object, *, secret: str | None = None, max_body_size: int = MAX_DEFAULT_BODY_SIZE, handle_in_background: bool = True, endpoint_type: str | None = None)

Bases: object

Parse and dispatch webhook requests without depending on an HTTP server.

async handle(bot: Bot[object], body: bytes, headers: Mapping[str, str]) → WebhookResult

Verify, parse and dispatch one webhook request.

Malformed JSON and oversized bodies are rejected. Authentication failures return the same 404 used by route failures so callers cannot distinguish a secret oracle from an absent endpoint.

async shutdown() → None

Wait for all background update tasks before server shutdown.

class peyk.webhook.WebhookRegistrar(*args, **kwargs)

Bases: Protocol

Register and remove a webhook for one platform.

async install(bot: Bot[object], url: str, secret: str, options: WebhookOptions) → None

Provides the install operation for the peyk integration.

Parameters:

• bot – Value used by this operation.

• url – Value used by this operation.

• secret – Value used by this operation.

• options – Value used by this operation.

async uninstall(bot: Bot[object]) → None

Provides the uninstall operation for the peyk integration.

Parameters:

bot – Value used by this operation.

class peyk.webhook.WebhookResult(status: int, body: bytes | None = None)

Bases: object

HTTP result returned by a webhook processor.

body: bytes | None = None

status: int

peyk.webhook.constant_time_compare(a: str | None, b: str | None) → bool

Compare two non-empty strings without ordinary early-exit timing.

peyk.webhook.registrar_for(bot: Bot[object]) → WebhookRegistrar

Return the audited registrar for bot.platform.

peyk.webhook.verify_telegram_secret_header(headers: Mapping[str, str], secret_token: str) → bool

Validate Telegram’s documented webhook secret header.

FSM

Finite-state conversation support for Peyk.

class peyk.fsm.BaseEventIsolation

Bases: object

Serialize handlers that share one StorageKey.

async close() → None

Close isolation resources.

lock(key: StorageKey) → AsyncIterator[None]

Return an async context manager for the event lock.

class peyk.fsm.BaseStorage(*, key_builder: KeyBuilder | None = None)

Bases: ABC

Abstract asynchronous FSM storage keyed by StorageKey.

abstractmethod async close() → None

Performs the close operation for the FSM client.

abstractmethod async get_data(key: StorageKey | str) → dict[str, object]

Retrieves data from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

abstractmethod async get_state(key: StorageKey | str) → str | None

Retrieves state from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

abstractmethod async set_data(key: StorageKey | str, data: Mapping[str, object]) → None

Updates data through the FSM API.

Parameters:

• key – Value used by this operation.

• data – Value used by this operation.

abstractmethod async set_state(key: StorageKey | str, state: str | None) → None

Updates state through the FSM API.

Parameters:

• key – Value used by this operation.

• state – Value used by this operation.

abstractmethod async update_data(key: StorageKey | str, data: Mapping[str, object] | None = None, **kwargs: object) → dict[str, object]

Performs the update data operation for the FSM client.

Parameters:

• key – Value used by this operation.

• data – Value used by this operation.

Returns:

Result produced by the FSM operation.

peyk.fsm.build_storage_key(event: object, *, platform: str, bot_id: int | str, strategy: str | FSMStrategy = FSMStrategy.USER_IN_CHAT, destiny: str = 'default') → StorageKey

Build a platform/bot-aware key from a normalized event.

peyk.fsm.conversation_key(event: object, *, platform: str | None = None, bot_id: int | str | None = None) → StorageKey

Return a compatibility StorageKey for an event.

The old form inferred the platform from event.raw and had no bot ID. That form is deprecated; dispatcher integrations must pass both values.

class peyk.fsm.DefaultKeyBuilder(prefix: str = 'peyk', separator: str = ':', with_bot_id: bool = True, with_destiny: bool = True)

Bases: object

Build aiogram-shaped FSM keys with platform and bot namespaces.

The default is equivalent to aiogram’s with_bot_id=True plus the additional platform segment required by Peyk’s multi-platform contract.

prefix: str

separator: str

with_bot_id: bool

with_destiny: bool

build(key: StorageKey, part: str | None = None) → str

Performs the build operation for the FSM client.

Parameters:

• key – Value used by this operation.

• part – Value used by this operation.

Returns:

Result produced by the FSM operation.

class peyk.fsm.DisabledEventIsolation

Bases: BaseEventIsolation

Do not serialize events.

lock(key: StorageKey) → AsyncIterator[None]

Performs the lock operation for the FSM client.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

class peyk.fsm.FSMContext(storage: BaseStorage, key: StorageKey)

class peyk.fsm.FSMContext(event: object, storage: BaseStorage | LegacyStringStorage, *, platform: str | None = None, bot_id: int | str | None = None)

Bases: object

Read and mutate state/data for one StorageKey.

async clear() → None

Clear both state and conversation data.

classmethod for_event(event: object, storage: BaseStorage, *, platform: str, bot_id: int | str, strategy: str | FSMStrategy = FSMStrategy.USER_IN_CHAT, destiny: str = 'default') → FSMContext

Create a context for a normalized event using explicit identity data.

async get_data() → dict[str, object]

Return a copy of conversation data.

async get_state() → str | None

Return the current state.

async get_value(key: str, default: object = None) → object

Return one data value, or default when absent.

async set_data(data: Mapping[str, object]) → None

Replace conversation data.

async set_state(state: State | str | None) → None

Set or clear the current state.

async update_data(data: Mapping[str, object] | None = None, **kwargs: object) → dict[str, object]

Merge mapping and keyword values into conversation data.

class peyk.fsm.FSMStrategy(*values)

Bases: str, Enum

Select which event identities share one FSM conversation.

USER_IN_CHAT = 'USER_IN_CHAT'

CHAT = 'CHAT'

GLOBAL_USER = 'GLOBAL_USER'

class peyk.fsm.KeyBuilder(*args, **kwargs)

Bases: Protocol

Build backend keys from a StorageKey.

build(key: StorageKey, part: str | None = None) → str

Performs the build operation for the FSM client.

Parameters:

• key – Value used by this operation.

• part – Value used by this operation.

Returns:

Result produced by the FSM operation.

class peyk.fsm.LegacyStorageAdapter(storage: LegacyStringStorage, *, key_builder: KeyBuilder | None = None)

Bases: BaseStorage

Adapt the pre-Phase-8 string-key storage API to StorageKey.

async close() → None

Performs the close operation for the FSM client.

async get_data(key: StorageKey | str) → dict[str, object]

Retrieves data from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

async get_state(key: StorageKey | str) → str | None

Retrieves state from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

async set_data(key: StorageKey | str, data: Mapping[str, object]) → None

Updates data through the FSM API.

Parameters:

• key – Value used by this operation.

• data – Value used by this operation.

async set_state(key: StorageKey | str, state: str | None) → None

Updates state through the FSM API.

Parameters:

• key – Value used by this operation.

• state – Value used by this operation.

async update_data(key: StorageKey | str, data: Mapping[str, object] | None = None, **kwargs: object) → dict[str, object]

Performs the update data operation for the FSM client.

Parameters:

• key – Value used by this operation.

• data – Value used by this operation.

Returns:

Result produced by the FSM operation.

class peyk.fsm.LegacyStringStorage(*args, **kwargs)

Bases: Protocol

Protocol for the pre-Phase-8 string-key storage API.

async close() → None

Performs the close operation for the FSM client.

async get_data(key: str) → dict[str, object]

Retrieves data from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

async get_state(key: str) → str | None

Retrieves state from the FSM API.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

async set_data(key: str, data: Mapping[str, object]) → None

Updates data through the FSM API.

Parameters:

• key – Value used by this operation.

• data – Value used by this operation.

async set_state(key: str, state: str | None) → None

Updates state through the FSM API.

Parameters:

• key – Value used by this operation.

• state – Value used by this operation.

async update_data(key: str, **kwargs: object) → dict[str, object]

Performs the update data operation for the FSM client.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

class peyk.fsm.MemoryStorage(*, key_builder: KeyBuilder | None = None)

Bases: BaseStorage

Store FSM state and data in process memory.

async close() → None

Release in-memory data and mark the backend closed.

async get_data(key: StorageKey | str) → dict[str, object]

Return a defensive copy of conversation data.

async get_state(key: StorageKey | str) → str | None

Return the stored state for key.

async set_data(key: StorageKey | str, data: Mapping[str, object]) → None

Replace conversation data with a defensive copy.

async set_state(key: StorageKey | str, state: str | None) → None

Set or clear the stored state.

async update_data(key: StorageKey | str, data: Mapping[str, object] | None = None, **kwargs: object) → dict[str, object]

Merge mapping and keyword values into conversation data.

class peyk.fsm.RedisEventIsolation(redis: RedisLockClient, *, key_builder: KeyBuilder | None = None, lock_timeout: float | None = 30.0, blocking_timeout: float | None = None)

Bases: BaseEventIsolation

Serialize events using Redis distributed locks.

async close() → None

Performs the close operation for the FSM client.

lock(key: StorageKey) → AsyncIterator[None]

Performs the lock operation for the FSM client.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

class peyk.fsm.RedisStorage(redis: AsyncRedisClient | None = None, *, state_ttl: int | float | None = None, data_ttl: int | float | None = None, key_builder: KeyBuilder | None = None, url: str | None = None, prefix: str | None = None)

Bases: BaseStorage

Persist FSM state/data in Redis with independent optional TTLs.

async close() → None

Close the underlying Redis client when supported.

async get_data(key: StorageKey | str) → dict[str, object]

Decode JSON conversation data from Redis.

async get_state(key: StorageKey | str) → str | None

Return a stored state string, if present.

async set_data(key: StorageKey | str, data: Mapping[str, object]) → None

Replace JSON conversation data and apply data_ttl.

async set_state(key: StorageKey | str, state: str | None) → None

Set or clear state, applying state_ttl when configured.

async update_data(key: StorageKey | str, data: Mapping[str, object] | None = None, **kwargs: object) → dict[str, object]

Read, merge and persist conversation data.

class peyk.fsm.SimpleEventIsolation

Bases: BaseEventIsolation

Serialize by key and evict idle locks after release.

async close() → None

Performs the close operation for the FSM client.

lock(key: StorageKey) → AsyncIterator[None]

Performs the lock operation for the FSM client.

Parameters:

key – Value used by this operation.

Returns:

Result produced by the FSM operation.

class peyk.fsm.State(state: str | None = None, group_name: str | None = None)

Bases: object

Describe one named FSM state, optionally with an explicit group name.

property group: type[StatesGroup]

Return the owning state-group class.

set_parent(group: type[StatesGroup]) → None

Bind this state to a StatesGroup class.

property state: str | None

Return the fully-qualified state name.

class peyk.fsm.StatesGroup

Bases: object

Base class for declarative FSM state groups.

classmethod get_root() → type[StatesGroup]

Return the root group for a nested group.

class peyk.fsm.StorageKey(platform: str, bot_id: int | str, chat_id: int | str | None, user_id: int | str | None, destiny: str = 'default')

Bases: object

Identify one FSM namespace without embedding a bot token.

platform is deliberately part of the identity so equal numeric IDs on Telegram and Bale cannot share state. bot_id isolates two bots on the same platform.

platform: str

bot_id: int | str

chat_id: int | str | None

user_id: int | str | None

destiny: str

Flags

class peyk.flags.FlagDecorator(name: str, value: object)

Bases: object

FlagDecorator defines a public API type for peyk.

name: str

value: object

class peyk.flags.Flags

Bases: object

Factory for handler flags used by dispatcher middleware.

peyk.flags.check_flags(flags: Mapping[str, object], required: Mapping[str, object] | None = None) → bool

Check that every requested flag exists and matches its required value.

peyk.flags.get_flag(handler: object, name: str, default: object = None) → object

Return one registered handler flag.

Utilities

class peyk.utils.I18n(path: str | Path, default_locale: str = 'fa', domain: str = 'messages')

Bases: object

Load GNU gettext catalogs and provide aiogram-style translation helpers.

gettext(message: str, locale: str | None = None) → str

Performs the gettext operation for the utility client.

Parameters:

• message – Value used by this operation.

• locale – Locale used for translation lookup.

Returns:

Result produced by the utility operation.

lazy_gettext(message: str, locale: str | None = None) → LazyText

Performs the lazy gettext operation for the utility client.

Parameters:

• message – Value used by this operation.

• locale – Locale used for translation lookup.

Returns:

Result produced by the utility operation.

ngettext(singular: str, plural: str, n: int, locale: str | None = None) → str

Performs the ngettext operation for the utility client.

Parameters:

• singular – Value used by this operation.

• plural – Value used by this operation.

• n – Value used by this operation.

• locale – Locale used for translation lookup.

Returns:

Result produced by the utility operation.

set_locale(locale: str) → None

Updates locale through the utility API.

Parameters:

locale – Locale used for translation lookup.

class peyk.utils.I18nMiddleware(i18n: I18n, get_locale: Callable[[object, Mapping[str, object]], str] | None = None)

Bases: object

Inject i18n, locale and translation callables into dispatcher DI.

class peyk.utils.LazyText(resolver: Callable[[], str])

Bases: object

LazyText provides the utility API surface used by peyk.

class peyk.utils.MediaGroupBuilder(items: list[~peyk.utils.media_group.T] = <factory>)

Bases: Generic[T]

Collect media items for a platform-supported media group.

add(media: T) → MediaGroupBuilder[T]

Performs the add operation for the utility client.

Parameters:

media – Value used by this operation.

Returns:

Result produced by the utility operation.

build() → list[T]

Performs the build operation for the utility client.

Returns:

Result produced by the utility operation.

items: list[T]

peyk.utils.check_webapp_signature(init_data: str, bot_token: str) → bool

Validate Telegram Web App init data according to the documented HMAC scheme.

peyk.utils.safe_parse_webapp_init_data(init_data: str, bot_token: str) → dict[str, str]

Validate and parse Telegram Web App init data.

peyk.utils.encode_payload(payload: str | bytes) → str

Encode UTF-8 payload using unpadded URL-safe base64.

peyk.utils.decode_payload(payload: str) → str

Decode an unpadded URL-safe base64 payload as UTF-8.

async peyk.utils.create_start_link(bot: object, payload: str | bytes) → str

Create the confirmed Telegram Bot API start deep link.