telebot_py.kernel

Application kernel: lifecycle, context, dispatcher, polling, webhook (T019-T022, T052).

telebot_py.kernel.app

Application kernel: lifecycle state machine, dispatch fan-out, runners (T021).

class telebot_py.kernel.app.Application(bot: Bot, *, post_init: Callable[[Application], Awaitable[None]] | None = None, post_shutdown: Callable[[Application], Awaitable[None]] | None = None, concurrent_updates: int = 32, sleep: Callable[[float], Awaitable[None]] | None = None, persistence: BasePersistence | None = None, job_queue_enabled: bool = False)

Bases: object

The bot application kernel: bot client, handlers, and lifecycle.

Drives the STOPPED -> INITIALIZING -> RUNNING -> STOPPING -> STOPPED state machine, owns the dispatcher and the per-user/per-chat/bot data dicts, and fans incoming updates out as tasks bounded by a semaphore (concurrent_updates). Handler exceptions never escape dispatch; they are routed to the registered error handlers (FR-013).

Example

>>> app = Application.builder().token("123:ABC...").build()
>>> app.add_handler(CommandHandler("start", start_callback))
>>> app.run_polling()  # blocks until Ctrl+C
Variables:

  • bot – The Bot API client.

  • dispatcher – The dispatcher routing updates to handler groups.

  • persistence – The configured persistence backend, when any (SC-005).

  • state – Current lifecycle state.

  • job_queue – The scheduler when enabled via the builder, else None; started once the bot is ready and stopped with the lifecycle.

  • plugin_manager – Dispatches plugin response/error hooks around handler results and failures (FR-015).

  • user_data – Mutable per-user dicts, created lazily per user id.

  • chat_data – Mutable per-chat dicts, created lazily per chat id.

add_error_handler(callback: Callable[[Update | None, CallbackContext], Awaitable[None] | None]) → None

Register a callback invoked as callback(update, context) on handler errors.

The raised exception is available as context.error.

Parameters:

callback – Sync or async error handler.

add_handler(handler: BaseHandler, group: int = 0) → None

Register an update handler into a dispatch group.

Parameters:

  • handler – The handler to register.

  • group – Group number; lower groups run first.

Example

>>> app.add_handler(CommandHandler("start", callback), group=0)
add_plugin(plugin: Plugin, *, order: int = 0) → None

Install a plugin into the update pipeline (FR-015).

Its response hook wraps every successful handler result and its error hook observes handler failures, both in resolved hook order.

Example

>>> app.add_plugin(I18nPlugin(default_locale="en", locales=tables))
Parameters:

  • plugin – The plugin to register.

  • order – Relative hook position; lower values hook first.

Raises:

  • ValueError – If a plugin with the same name is already registered.

  • PluginOrderingError – If the dependency graph now has a cycle.

property bot_data: dict[Any, Any]

Mutable dict shared by every update this application processes.

classmethod builder() → ApplicationBuilder

Create a fluent ApplicationBuilder (python-telegram-bot parity).

Returns:

A fresh builder.

async initialize() → None

Initialize the application: validate the token, run post-init hooks.

The startup get_me call fails fast with InvalidTokenError on a rejected token. When a persistence backend is configured, bot/user/chat data and persistent conversation state are loaded here (SC-005). An enabled JobQueue is started once the bot is ready. On success the state becomes INITIALIZING and stays there until start().

Raises:
async process_update(update: Update) → None

Dispatch one update through the handler groups, bounded by the semaphore.

Parameters:

update – The incoming update.

Returns:

None once every matching handler finished (or its failure was routed to the error handlers).

Raises:

ApplicationError – If the application is not RUNNING.

remove_handler(handler: BaseHandler, group: int = 0) → None

Unregister a handler.

Parameters:

  • handler – The handler to remove.

  • group – Group the handler was registered in.

Raises:

ValueError – If the handler is not registered in that group.

remove_plugin(name: str) → None

Uninstall a plugin; its hooks stop firing before the next update.

Parameters:

name – Name of the registered plugin to remove.

Raises:

ValueError – If no plugin with that name is registered.

run_polling(*, allowed_updates: Sequence[str] | None = None, drop_pending_updates: bool = False, poll_interval: float = 0.0, timeout: float = 10, stop_signals: Sequence[int] = (Signals.SIGINT, Signals.SIGTERM), close_loop: bool = True) → None

Run the full polling lifecycle, blocking until a stop signal arrives.

Creates an event loop, installs the stop signal handlers (Ctrl+C by default), then runs initialize -> start -> getUpdates loop -> stop -> shutdown. Only one run mode is ever allowed per application: a second run_polling/run_webhook raises.

Example

>>> app.run_polling(drop_pending_updates=True)  # blocks until Ctrl+C
Parameters:

  • allowed_updates – Update types to receive.

  • drop_pending_updates – Discard updates queued while the bot was offline before starting to poll.

  • poll_interval – Extra delay between empty poll responses.

  • timeout – Long-poll timeout in seconds handed to Telegram.

  • stop_signals – Signals that trigger a graceful stop.

  • close_loop – Whether to close the created loop on exit.

Raises:

  • ApplicationError – If the application already ran, is not STOPPED, or this is called from a running event loop.

  • InvalidTokenError – If Telegram rejects the token at startup.

run_webhook(*, listen: str = '127.0.0.1', port: int = 8443, url_path: str = '', secret_token: str | None = None, stop_signals: Sequence[int] = (Signals.SIGINT, Signals.SIGTERM), close_loop: bool = True) → None

Run the full webhook lifecycle, blocking until a stop signal arrives.

Creates an event loop, runs initialize -> start -> webhook server -> stop -> shutdown, and serves deliveries on http://listen:port at url_path until a stop signal arrives. Only one run mode is ever allowed per application: a second run_polling/run_webhook raises. TLS termination is the operator’s job: bind localhost behind a reverse proxy. Registration at Telegram is likewise the operator’s: run_webhook never calls set_webhook/delete_webhook (packages/node parity).

Example

>>> app.run_webhook(url_path="hook", secret_token="s3cret")
Parameters:

  • listen – Interface to bind the webhook server to.

  • port – Port to serve the webhook on.

  • url_path – URL path Telegram should POST updates to.

  • secret_token – Token Telegram must echo in X-Telegram-Bot-Api-Secret-Token; deliveries without it get a 401.

  • stop_signals – Signals that trigger a graceful stop.

  • close_loop – Whether to close the created loop on exit.

Raises:

  • ApplicationError – If the application already ran, is not STOPPED, or this is called from a running event loop.

  • InvalidTokenError – If Telegram rejects the token at startup.

  • OSError – If the listen address/port cannot be bound.

async shutdown() → None

Shut down: run post-shutdown hooks, close persistence and the Bot client.

Raises:

ApplicationError – If never initialized, not stopped yet, or already shut down.

async start() → None

Enter the RUNNING state so updates can be processed.

Raises:

ApplicationError – If initialize() has not succeeded yet.

async stop() → None

Stop processing: signal the run loop and drain in-flight updates.

The enabled JobQueue is stopped before the drain so no new jobs fire while in-flight updates finish.

Raises:

ApplicationError – If the application is not RUNNING.

telebot_py.kernel.builder

Fluent ApplicationBuilder with fail-fast validation at build() (T021, FR-014).

class telebot_py.kernel.builder.ApplicationBuilder

Bases: object

Fluent builder assembling a configured Application.

Collects the bot token, an optional transport override (the offline test seam), lifecycle hooks, and the concurrency bound, validating the combination at build() rather than deep inside the runtime (FR-014).

Example

>>> app = ApplicationBuilder().token("123:ABC...").build()
build() → Application

Validate the collected configuration and construct the Application.

Returns:

The configured application in the STOPPED state.

Raises:

ValueError – If no (or an empty) token was set, or any combination of options is invalid (FR-014).

concurrent_updates(count: int) → ApplicationBuilder

Set the maximum number of updates dispatched concurrently.

Parameters:

count – Positive concurrency bound; defaults to 32 when unset.

Returns:

This builder for chaining.

Raises:

ValueError – If count is not a positive integer.

job_queue(enabled: bool = True) → ApplicationBuilder

Enable or disable the application’s JobQueue (FR-008).

When enabled, build() attaches a fresh JobQueue that initialize() starts once the bot is ready and stop()/shutdown() stop cleanly. Without this call the application has no job queue and context.job_queue stays None.

Parameters:

enabled – Whether the built application owns a JobQueue.

Returns:

This builder for chaining.

Raises:

ValueError – If enabled is not a bool, or the option was already set on this builder.

persistence(persistence: BasePersistence) → ApplicationBuilder

Attach the persistence backend for conversation and data durability.

The application loads bot/user/chat data and restores persistent conversation handlers during initialize(), writes everything back on stop(), and shuts the backend down in shutdown() (SC-005).

Parameters:

persistence – A BasePersistence instance.

Returns:

This builder for chaining.

Raises:

ValueError – If a persistence backend was already set.

post_init(callback: PostLifecycleCallback) → ApplicationBuilder

Register a coroutine run at the end of Application.initialize().

Parameters:

callback – Awaited with the application once initialization and the startup get_me check succeeded.

Returns:

This builder for chaining.

post_shutdown(callback: PostLifecycleCallback) → ApplicationBuilder

Register a coroutine run during Application.shutdown().

Parameters:

callback – Awaited with the application before the Bot client is closed.

Returns:

This builder for chaining.

sleep(sleep: Callable[[float], Awaitable[None]]) → ApplicationBuilder

Inject an async sleep used for retry and polling backoff (test seam).

Parameters:

sleep – Coroutine function accepting a delay in seconds.

Returns:

This builder for chaining.

token(token: str) → ApplicationBuilder

Set the bot token issued by BotFather.

Parameters:

token – Bot token string; validated non-empty at build time.

Returns:

This builder for chaining.

Raises:

ValueError – If a token was already set.

transport(transport: AsyncBaseTransport) → ApplicationBuilder

Inject a transport replacing only the wire layer (test seam).

Retry policy and error mapping stay active even with a custom transport, e.g. httpx.MockTransport serving canned responses.

Parameters:

transport – An httpx async transport instance.

Returns:

This builder for chaining.

telebot_py.kernel.builder.DEFAULT_CONCURRENT_UPDATES = 32

Default bound on concurrently dispatched updates (python-telegram-bot style).

telebot_py.kernel.context

Per-update callback context passed to handler and error callbacks (T019).

class telebot_py.kernel.context.CallbackContext(application: Application, update: Update)

Bases: object

Context object delivered as the second argument to handler callbacks.

Created fresh for every matched handler by the dispatcher. Carries the update being processed, shortcuts to its effective chat/user/message, the per-user, per-chat, and bot-wide mutable data dicts, and slots populated by specific handlers (args for commands, matches for regex patterns). Error handlers additionally find the raised exception in error.

Example

>>> async def handler(update, context):
...     await context.bot.send_message(context.effective_chat.id, "hi")
Variables:

  • application – The application processing the update.

  • bot – The Bot client used to call the Telegram API.

  • update – The update currently being processed.

  • user_data – Mutable dict for the effective user, None when the update carries no user. Shared across updates of the same user.

  • chat_data – Mutable dict for the effective chat, None when the update carries no chat. Shared across updates of the same chat.

  • bot_data – Mutable dict shared by every update this application sees.

  • job_queue – The application’s JobQueue when the builder enabled it (job_queue()), otherwise None.

  • matches – Regex matches collected by pattern-based handlers.

  • args – Command arguments collected by CommandHandler.

  • error – The exception being reported, set for error handlers only.

property effective_chat: Chat | None

The chat this update happened in, when known.

property effective_user: User | None

The user who triggered this update, when known.

classmethod from_update(update: Update, application: Application) → CallbackContext

Build a context for update (python-telegram-bot parity factory).

Parameters:

  • update – The update being processed.

  • application – The application whose bot and data dicts back the context.

Returns:

A fresh CallbackContext bound to the update.

property message: Message | None

The message carried by the update, regardless of payload type.

telebot_py.kernel.dispatcher

Update dispatcher: group-ordered routing and error handler fan-out (T020).

class telebot_py.kernel.dispatcher.Dispatcher(application: Application)

Bases: object

Routes updates to handler groups and failures to error handlers.

Groups are processed in ascending key order; within a group the first handler whose check_update returns something other than None or False wins and the remaining handlers of that group are skipped. A winning handler with block=True (the default) consumes the update: higher groups never see it. With block=False the callback is fired as a background task and higher groups still get their chance.

Exceptions raised by check_update or a handler callback never escape process_update(); they are logged and routed to every registered error handler, and subsequent groups keep running (FR-013). Plugin hooks wrap the same stage: response hooks run around every successful handler result, error hooks observe failures after the error handlers ran.

Example

>>> dispatcher.add_handler(handler, group=0)
>>> await dispatcher.process_update(update)
Variables:

  • application – The application owning this dispatcher.

  • bot – Shortcut to the application’s Bot client.

add_error_handler(callback: Callable[[Update | None, CallbackContext], Awaitable[None] | None]) → None

Register a callback invoked with (update, context) on handler errors.

Parameters:

callback – Sync or async callable; context.error carries the raised exception.

Example

>>> async def on_error(update, context):
...     logging.error("update failed", exc_info=context.error)
>>> dispatcher.add_error_handler(on_error)
add_handler(handler: BaseHandler, group: int = 0) → None

Register a handler into a group (lower groups run first).

Parameters:

  • handler – The handler to register.

  • group – Group number; defaults to 0.

Example

>>> dispatcher.add_handler(CommandHandler("start", callback), group=0)
property background_tasks: set[Task[None]]

Snapshot of in-flight background tasks from non-blocking handlers.

property error_handlers: list[Callable[[Update | None, CallbackContext], Awaitable[None] | None]]

Registered error handler callbacks, in registration order.

property groups: list[int]

Group numbers in ascending dispatch order.

property handlers: dict[int, list[BaseHandler]]

Handler groups keyed by group number, registration order within.

async process_update(update: Update) → None

Dispatch one update through all handler groups; never raises.

Parameters:

update – The incoming update.

Returns:

None. Handler exceptions are routed to the error handlers instead of propagating (FR-013).

remove_error_handler(callback: Callable[[Update | None, CallbackContext], Awaitable[None] | None]) → None

Unregister an error handler; no-op when not registered.

Parameters:

callback – The previously registered callback.

remove_handler(handler: BaseHandler, group: int = 0) → None

Unregister a handler from a group.

Parameters:

  • handler – The handler to remove.

  • group – Group the handler was registered in.

Raises:

ValueError – If the handler is not registered in that group.

telebot_py.kernel.dispatcher.ErrorHandler

the update being processed (None for errors not tied to one) and a context whose error attribute carries the raised exception.

Type:

Error handler signature (python-telegram-bot parity)

alias of Callable[[Update | None, CallbackContext], Awaitable[None] | None]

telebot_py.kernel.lifecycle

Application lifecycle state machine (data-model.md section 2).

The kernel moves through STOPPED -> INITIALIZING -> RUNNING -> STOPPING -> STOPPED; transitions are enforced by Application raising ApplicationError on misuse.

class telebot_py.kernel.lifecycle.ApplicationState(*values)

Bases: Enum

Lifecycle state of an Application.

Variables:

  • STOPPED – Built but not initialized, or fully stopped after stop().

  • INITIALIZING – initialize() succeeded; start() has not run yet.

  • RUNNING – The application is processing updates.

  • STOPPING – stop() is draining in-flight updates.

INITIALIZING = 'initializing'
RUNNING = 'running'
STOPPED = 'stopped'
STOPPING = 'stopping'

telebot_py.kernel.polling

Long-polling getUpdates loop with offset management (T022).

telebot_py.kernel.polling.BACKOFF_BASE_SECONDS = 1.0

1s, 2s, 4s, … capped at 30s (mirrors the HTTP retry policy shape, FR-012).

Type:

Polling-loop backoff after the Bot’s own retries were exhausted

telebot_py.kernel.polling.install_stop_signal_handlers(loop: AbstractEventLoop, stop_event: Event, stop_signals: Sequence[int]) → list[Callable[[], None]]

Install handlers setting stop_event for each signal.

Prefers loop-level handlers; falls back to signal.signal on platforms where the loop does not support them (e.g. Windows proactor loops).

Parameters:

  • loop – The running event loop to install handlers on.

  • stop_event – Event set when a stop signal arrives.

  • stop_signals – Signal numbers to catch (e.g. SIGINT, SIGTERM).

Returns:

Cleanup callables restoring the previous handler state, in installation order.

async telebot_py.kernel.polling.poll_updates(*, bot: Bot, on_update: Callable[[Update], object], stop_event: Event, allowed_updates: Sequence[str] | None = None, drop_pending_updates: bool = False, timeout: float = 10, poll_interval: float = 0.0, sleep: Callable[[float], Awaitable[None]] | None = None) → None

Run the getUpdates loop until stop_event is set.

Confirms updates by advancing offset to the last seen update_id plus one. drop_pending_updates performs a flush call with offset=-1 first so Telegram discards everything queued while the bot was offline (python-telegram-bot parity). Transport and API failures are logged and retried with capped exponential backoff instead of crashing; an InvalidTokenError is fatal and re-raised immediately.

Example

>>> await poll_updates(bot=bot, on_update=spawn, stop_event=stop)
Parameters:

  • bot – Bot client whose get_updates feeds the loop.

  • on_update – Synchronous callable invoked once per fetched update; expected to schedule dispatch (e.g. as a task) and return immediately so updates do not block each other.

  • stop_event – Loop exits as soon as this event is set.

  • allowed_updates – Update types to receive, passed through to Telegram.

  • drop_pending_updates – Whether to discard updates queued while offline.

  • timeout – Long-poll timeout in seconds handed to Telegram.

  • poll_interval – Extra delay between requests when a batch was empty.

  • sleep – Injectable async sleep (tests); defaults to asyncio.sleep.

Returns:

None once the stop event is observed and the in-flight request completed.

telebot_py.kernel.runners

Run-mode lifecycles: blocking polling and webhook runners (T022, T052).

Module-level implementations behind Application.run_polling() and Application.run_webhook(), split out of app.py to keep the kernel file focused. They deliberately drive application internals through the full initialize -> start -> serve -> stop -> shutdown cycle.

telebot_py.kernel.runners.assert_can_run(app: Application, mode: str) → None

Fail fast on run-mode misuse before either runner starts.

Parameters:

  • app – The application about to run.

  • mode – Runner name ("polling" or "webhook") for error messages.

Raises:

ApplicationError – If the app already ran in any mode, is not STOPPED, or an event loop is already running in this thread.

async telebot_py.kernel.runners.polling_lifecycle(app: Application, *, allowed_updates: Sequence[str] | None = None, drop_pending_updates: bool = False, poll_interval: float = 0.0, timeout: float = 10, stop_signals: Sequence[int] = ()) → None

Implementation of Application._run_polling; see its docstring.

Parameters:

  • app – The application to run.

  • allowed_updates – Update types to receive.

  • drop_pending_updates – Discard updates queued while offline first.

  • poll_interval – Extra delay between empty poll responses.

  • timeout – Long-poll timeout in seconds handed to Telegram.

  • stop_signals – Signals that trigger a graceful stop.

telebot_py.kernel.runners.run_in_new_loop(coro: Coroutine[Any, Any, None], *, close_loop: bool) → None

Block on a lifecycle coroutine in a freshly created event loop.

Parameters:

  • coro – The run-mode lifecycle coroutine.

  • close_loop – Whether to close the created loop on exit.

async telebot_py.kernel.runners.webhook_lifecycle(app: Application, *, listen: str = '127.0.0.1', port: int = 8443, url_path: str = '', secret_token: str | None = None, stop_signals: Sequence[int] = ()) → None

Implementation of Application._run_webhook; see its docstring.

Webhook registration at Telegram is left to the operator (node parity); this lifecycle only serves deliveries until the stop event fires.

Parameters:

  • app – The application to run.

  • listen – Interface to bind the webhook server to.

  • port – Port to serve the webhook on; 0 picks a free ephemeral port.

  • url_path – URL path Telegram should POST updates to.

  • secret_token – Token deliveries must echo, or None to skip verification.

  • stop_signals – Signals that trigger a graceful stop.

telebot_py.kernel.webhook

Webhook HTTP server: asyncio streams plus a minimal HTTP/1.1 parser (T052).

Implements research R4: one POST route served by asyncio.start_server with no ASGI dependency. The parser understands a request line, headers, and a fixed Content-Length body; chunked, oversized, truncated, and malformed requests are rejected with an HTTP error status and never kill the server (quickstart V6). TLS termination is the operator’s responsibility — bind the server to localhost behind a reverse proxy in production.

Webhook registration at Telegram (set_webhook/delete_webhook) is deliberately left to the operator, mirroring packages/node’s runWebhook, which also only serves deliveries.

telebot_py.kernel.webhook.MAX_HEADER_COUNT = 100

Header-count guard rail so hostile requests cannot exhaust memory.

telebot_py.kernel.webhook.MAX_WEBHOOK_BODY_BYTES = 5242880

Maximum accepted size in bytes of an incoming webhook request body (parity with packages/node MAX_WEBHOOK_BODY_BYTES).

telebot_py.kernel.webhook.OnUpdate

Callback receiving each parsed update (e.g. Application._spawn_update).

alias of Callable[[Update], object]

telebot_py.kernel.webhook.REQUEST_TIMEOUT_SECONDS = 10.0

Seconds a client may take to deliver the request head and body.

telebot_py.kernel.webhook.SECRET_TOKEN_HEADER = 'x-telegram-bot-api-secret-token'

Header carrying the shared secret on every Telegram delivery.

exception telebot_py.kernel.webhook.WebhookRequestError(status: int, detail: str)

Bases: Exception

A request that must be answered with an HTTP error status.

Variables:

status – The HTTP status code to respond with.

class telebot_py.kernel.webhook.WebhookServer(*, listen: str, port: int, url_path: str, secret_token: str | None, on_update: Callable[[Update], object])

Bases: object

Single-route HTTP/1.1 endpoint feeding parsed updates to a callback.

Each connection is handled independently and closed after one response; routing, secret-token, and body errors produce 400/401/404/405 responses without ever stopping the server.

Example

>>> server = await serve_webhook(
...     listen="127.0.0.1", port=8443, url_path="/hook",
...     secret_token="s3cret", on_update=app._spawn_update,
... )
Variables:

  • url_path – The path Telegram must POST updates to.

  • secret_token – The expected secret token, or None when verification is disabled.

async close() → None

Stop accepting connections and cancel any in-flight handlers.

property port: int

The port actually bound (differs from the request when it was 0).

async serve() → None

Bind and start accepting connections.

Raises:

OSError – If the address/port cannot be bound.

telebot_py.kernel.webhook.is_secret_token_valid(received: str | None, expected: str) → bool

Compare a received secret token against the configured one.

Uses a constant-time comparison so response timing cannot leak how many leading characters matched (parity with packages/node).

Parameters:

  • received – Raw X-Telegram-Bot-Api-Secret-Token header value.

  • expected – The secret token configured for this webhook.

Returns:

True when received matches expected.

async telebot_py.kernel.webhook.serve_webhook(*, listen: str, port: int, url_path: str, secret_token: str | None, on_update: Callable[[Update], object]) → WebhookServer

Create a WebhookServer and start listening.

Parameters:

  • listen – Interface address to bind to.

  • port – Port to bind; 0 picks a free ephemeral port.

  • url_path – Route accepting POSTed updates.

  • secret_token – Shared secret for delivery verification, or None.

  • on_update – Callback invoked once per accepted, parsed update.

Returns:

The listening server; read its port for the actually bound port.

Raises:

OSError – If the address/port cannot be bound.