### Multi-Bot Setup Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/base-webhook-engine.md Example demonstrating the setup for a TokenEngine, suitable for managing multiple bots. It configures a route with a dynamic bot token parameter. ```python from aiogram_webhook import TokenEngine, Route, BotTokenParam engine = TokenEngine( dispatcher=dispatcher, web=FastAPIAdapter(), route=Route( base_url="https://example.com", path="/webhook/{bot_token}", params={"bot_token": BotTokenParam()} ) ) engine.register(app) # When POST /webhook/456:XYZ is called: # 1. route.match() extracts bot_token="456:XYZ" # 2. _resolve_target() returns Target(bot_id=456, bot_token="456:XYZ") # 3. _resolve_bot() creates/retrieves bot with token "456:XYZ" # 4. Update is fed to dispatcher ``` -------------------------------- ### Complete Security Setup Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/security.md Demonstrates a complete setup for aiogram-webhook using FastAPI, including security configurations with IP checks and static secret tokens. ```python from aiogram_webhook import ( SingleBotEngine, FastAPIAdapter, Route, ) from aiogram_webhook.security import Security, StaticSecretToken from aiogram_webhook.security.checks.ip import IPCheck from fastapi import FastAPI from aiogram import Bot, Dispatcher app = FastAPI() dispatcher = Dispatcher() bot = Bot(token="123:ABC-token") ``` -------------------------------- ### Single-Bot Setup Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/base-webhook-engine.md Example demonstrating how to set up and register a SingleBotEngine with FastAPI. This setup is used when your application manages a single Telegram bot. ```python from aiogram import Bot, Dispatcher from aiogram_webhook import SingleBotEngine, FastAPIAdapter, Route from fastapi import FastAPI app = FastAPI() dispatcher = Dispatcher() bot = Bot(token="123:ABC") engine = SingleBotEngine( dispatcher=dispatcher, bot=bot, web=FastAPIAdapter(), route=Route(base_url="https://example.com", path="/webhook") ) engine.register(app) @app.on_event("startup") async def startup(): await engine.set_webhook() # When POST /webhook is called: # 1. route.match() extracts params (none in this case) # 2. _resolve_target() returns Target(bot_id=123, bot_token="123:ABC") # 3. _resolve_bot() returns self.bot # 4. Update is fed to dispatcher ``` -------------------------------- ### Complete Single-Bot Setup with FastAPI Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/README.md A full example demonstrating how to set up a single bot engine with FastAPI, including bot initialization, security configuration, and webhook event handling for startup and shutdown. ```python from fastapi import FastAPI from aiogram import Bot, Dispatcher from aiogram_webhook import SingleBotEngine, FastAPIAdapter, Route from aiogram_webhook.security import Security, StaticSecretToken app = FastAPI() dispatcher = Dispatcher() bot = Bot(token="123:ABC-token") security = Security(secret_token=StaticSecretToken("my_secret_token")) engine = SingleBotEngine( dispatcher=dispatcher, bot=bot, web=FastAPIAdapter(), route=Route(base_url="https://example.com", path="/webhook"), security=security ) engine.register(app) @app.on_event("startup") async def startup(): await engine.set_webhook() @app.on_event("shutdown") async def shutdown(): if bot.session: await bot.session.close() ``` -------------------------------- ### BotConfig Initialization Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Shows how to instantiate BotConfig, providing an AiohttpSession for bot communication. ```python from aiogram_webhook import BotConfig from aiogram.client.session.aiohttp import AiohttpSession config = BotConfig(session=AiohttpSession()) ``` -------------------------------- ### Production Setup with aiogram-webhook and FastAPI Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/configuration.md A comprehensive example demonstrating the integration of aiogram-webhook with FastAPI for a production environment. It includes bot configuration, webhook setup, security, and routing. ```python from fastapi import FastAPI from aiogram import Bot, Dispatcher from aiogram.client.default import DefaultBotProperties from aiogram.client.session.aiohttp import AiohttpSession from aiogram_webhook import ( SingleBotEngine, FastAPIAdapter, Route, WebhookConfig, BotConfig, ) from aiogram_webhook.security import Security, StaticSecretToken from aiogram_webhook.security.checks.ip import IPCheck # Create FastAPI app app = FastAPI() # Configure bot with shared session and default properties shared_session = AiohttpSession() bot_config = BotConfig( session=shared_session, default=DefaultBotProperties(parse_mode="HTML") ) # Create bot and dispatcher bot = Bot(token="123:ABC-token") dispatcher = Dispatcher() # Configure webhook webhook_config = WebhookConfig( max_connections=50, allowed_updates=["message", "callback_query", "inline_query"], drop_pending_updates=False ) # Configure security security = Security( IPCheck(include_default=True), secret_token=StaticSecretToken("my_webhook_secret_token_xyz") ) # Configure route route = Route( base_url="https://bot-api.example.com", path="/webhook" ) # Create and register engine engine = SingleBotEngine( dispatcher=dispatcher, bot=bot, web=FastAPIAdapter(), route=route, security=security, handle_in_background=True, shutdown_timeout=15.0 ) engine.register(app) @app.on_event("startup") async def startup(): await engine.set_webhook(webhook_config=webhook_config) @app.on_event("shutdown") async def shutdown(): await bot.session.close() ``` -------------------------------- ### FastAPI Web Adapter Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/web-adapters.md Demonstrates how to integrate aiogram-webhook with a FastAPI application using FastAPIAdapter. Ensure FastAPI and aiogram are installed. ```python from fastapi import FastAPI from aiogram_webhook import FastAPIAdapter, SingleBotEngine, Route from aiogram import Bot, Dispatcher app = FastAPI() dispatcher = Dispatcher() bot = Bot(token="123:ABC") adapter = FastAPIAdapter() engine = SingleBotEngine( dispatcher=dispatcher, bot=bot, web=adapter, route=Route(base_url="https://example.com", path="/webhook") ) engine.register(app) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) ``` -------------------------------- ### Example Usage: Simple single-bot webhook Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/route.md Demonstrates the basic setup of a Route for a single-bot webhook with a fixed base URL and a simple path. ```APIDOC ## Example Usage: Simple single-bot webhook ```python from aiogram_webhook import Route route = Route( base_url="https://example.com", path="/webhook" ) # URL: https://example.com/webhook ``` ``` -------------------------------- ### Complete Multi-Bot Setup with aiohttp Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/README.md Example for setting up a token engine to manage multiple bots with aiohttp. It shows dynamic bot addition and removal, including webhook deletion. ```python from aiohttp.web import Application from aiogram import Dispatcher from aiogram_webhook import TokenEngine, AiohttpAdapter, Route, BotTokenParam, BotConfig from aiogram.client.session.aiohttp import AiohttpSession app = Application() dispatcher = Dispatcher() # Shared session for all bots session = AiohttpSession() bot_config = BotConfig(session=session) engine = TokenEngine( dispatcher=dispatcher, web=AioiohttpAdapter(), route=Route( base_url="https://example.com", path="/webhook/{bot_token}", params={"bot_token": BotTokenParam()} ), bot_config=bot_config ) engine.register(app) # Add bots dynamically bot1 = await engine.add_bot("123:ABC-token") bot2 = await engine.add_bot("456:XYZ-token") # Remove bot await engine.remove_bot(bot1.id, delete_webhook=True) ``` -------------------------------- ### FastAPI Adapter Usage Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Demonstrates how to initialize and use the FastAPIAdapter. ```python from aiogram_webhook import FastAPIAdapter from fastapi import FastAPI app = FastAPI() adapter = FastAPIAdapter() ``` -------------------------------- ### Setup SingleBotEngine Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/engines/single-bot-engine.md Initializes the SingleBotEngine with a dispatcher, bot instance, web adapter, and route configuration. This is the basic setup for a single bot application. ```python from aiogram import Bot, Dispatcher from aiogram_webhook import FastAPIAdapter, SingleBotEngine from aiogram_webhook.route import Route dispatcher = Dispatcher() bot = Bot("BOT_TOKEN") engine = SingleBotEngine( dispatcher, bot, web=FastAPIAdapter(), route=Route(base_url="https://example.com", path="/webhook"), ) ``` -------------------------------- ### SingleBotEngine Initialization Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Illustrates how to create an instance of SingleBotEngine, providing the necessary dispatcher, bot, web adapter, and route. ```python from aiogram_webhook import SingleBotEngine engine = SingleBotEngine( dispatcher=dispatcher, bot=bot, web=adapter, route=route ) ``` -------------------------------- ### Aiohttp Web Adapter Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/web-adapters.md Shows how to set up aiogram-webhook with an aiohttp application using AiohttpAdapter. This example includes setting the webhook on startup. ```python from aiohttp.web import Application, run_app from aiogram_webhook import AiohttpAdapter, SingleBotEngine, Route from aiogram import Bot, Dispatcher import asyncio async def main(): app = Application() dispatcher = Dispatcher() bot = Bot(token="123:ABC") adapter = AiohttpAdapter() engine = SingleBotEngine( dispatcher=dispatcher, bot=bot, web=adapter, route=Route(base_url="https://example.com", path="/webhook") ) engine.register(app) # Set webhook on startup runner = web.AppRunner(app) await runner.setup() await engine.set_webhook() return app if __name__ == "__main__": app = asyncio.run(main()) run_app(app, host="0.0.0.0", port=8000) ``` -------------------------------- ### Security Verify Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/security.md Example of setting up and using the Security class with IP checks and a static secret token for request verification. ```python from aiogram_webhook.security import Security, IPCheck, StaticSecretToken security = Security( IPCheck(include_default=True), secret_token=StaticSecretToken("my_secret_token_123") ) # Raises SecretTokenError or SecurityCheckError if verification fails await security.verify(target=target, request=request, route_params=route_params) ``` -------------------------------- ### aiohttp Adapter Usage Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Demonstrates how to initialize and use the AiohttpAdapter. ```python from aiogram_webhook import AiohttpAdapter from aiohttp.web import Application app = Application() adapter = AiohttpAdapter() ``` -------------------------------- ### Complex Route with All Features Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/query-parameters.md Illustrates a comprehensive route configuration including path parameters, dynamic query parameters, and strict query validation. This example shows how to build a URL for a multi-bot setup with versioning and client identification. ```APIDOC ## Complex Route with All Features ### Description Defines a complex route for a multi-bot system, incorporating path parameters for bot identification and token, and query parameters for API version, client type, and dynamic bot ID referencing. Strict query validation is enabled. ### Method Not applicable (configuration example) ### Endpoint `/bots/{bot_id}/webhook/{bot_token}` ### Parameters #### Path Parameters - **bot_id** (string) - Required - The unique identifier for the bot. - **bot_token** (string) - Required - The authentication token for the bot. #### Query Parameters - **api_version** (string) - Required - Specifies the API version (e.g., "v7"). - **client** (string) - Required - Identifies the client type (e.g., "telegram"). - **bot_id** (string) - Required - References the bot ID from the path parameter. - **formats** (list of strings) - Required - Allowed data formats (e.g., ["json", "protobuf"]). ### Configuration Example ```python from aiogram_webhook import Route, BotIdParam, BotTokenParam, Const, Ref route = Route( base_url="https://api.example.com:8443", path="/bots/{bot_id}/webhook/{bot_token}", params={ "bot_id": BotIdParam(), "bot_token": BotTokenParam() }, query={ "api_version": Const("v7"), "client": Const("telegram"), "bot_id": Ref("bot_id"), "formats": ["json", "protobuf"] }, strict_query=True ) ``` ### Generated URL Example `https://api.example.com:8443/bots/123456789/webhook/123456789:ABCtoken/?api_version=v7&client=telegram&bot_id=123456789&formats=json&formats=protobuf` ``` -------------------------------- ### Basic AiohttpAdapter Setup Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/web/aiohttp.md This snippet demonstrates the basic setup of AiohttpAdapter with SingleBotEngine, including bot and dispatcher initialization, route configuration, and webhook registration within an aiohttp application lifecycle. ```python from aiohttp import web from aiogram import Bot, Dispatcher from aiogram_webhook import AiohttpAdapter, SingleBotEngine from aiogram_webhook.route import Route dispatcher = Dispatcher() bot = Bot("BOT_TOKEN") engine = SingleBotEngine( dispatcher, bot, web=AiohttpAdapter(), route=Route(base_url="https://example.com", path="/webhook"), ) async def set_webhook(app: web.Application) -> None: await engine.set_webhook() app = web.Application() app.on_startup.append(set_webhook) engine.register(app) ``` -------------------------------- ### Route Path Property Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/route.md Demonstrates how to get the path template string from a Route instance. This is useful for registering routes with web frameworks. ```python route = Route( base_url="https://example.com", path="/webhook/{bot_id}", params={"bot_id": BotIdParam()} ) print(route.path) # Output: /webhook/{bot_id} ``` -------------------------------- ### WebhookConfig Initialization Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Demonstrates initializing WebhookConfig with a specific number of maximum connections and its subsequent use with an engine. ```python from aiogram_webhook import WebhookConfig config = WebhookConfig(max_connections=50) await engine.set_webhook(webhook_config=config) ``` -------------------------------- ### Multiple Query Values Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/query-parameters.md A complete example demonstrating a Route configured to accept multiple values for a single query parameter ('type'). ```python from aiogram_webhook import Route, Const route = Route( base_url="https://example.com", path="/webhook", query={ "type": ["message", "callback_query", "inline_query"] } ) ``` -------------------------------- ### Engine Initialization with Custom Adapter Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/web/custom.md Example of how to initialize the SingleBotEngine with a custom adapter. The engine and route configuration remain the same, only the `web` parameter changes. ```python engine = SingleBotEngine( dispatcher, bot, web=MyAdapter(), route=route, security=security, ) engine.register(app) ``` -------------------------------- ### Static and Dynamic Query Parameters Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/query-parameters.md An example showcasing a Route with both a static query parameter ('api') and a dynamic one ('bot_id') derived from the path. The generated URL includes both. ```python from aiogram_webhook import Route, BotIdParam, Const, Ref route = Route( base_url="https://example.com", path="/webhook/{bot_id}", params={"bot_id": BotIdParam()}, query={ "api": Const("v1"), "bot_id": Ref("bot_id") } ) # URL: https://example.com/webhook/123456789?api=v1&bot_id=123456789 # Requests must match this exact query string ``` -------------------------------- ### StaticSecretToken Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/security.md Example of creating a StaticSecretToken with a valid token and handling potential ValueErrors for invalid token formats. ```python from aiogram_webhook.security.secret_token import StaticSecretToken # Create with a valid token token = StaticSecretToken("my_secret_abc123-xyz") # Invalid tokens raise ValueError try: token = StaticSecretToken("invalid token with spaces") # Raises ValueError except ValueError as e: print(f"Invalid token: {e}") ``` -------------------------------- ### Usage Example for Top-Level Package Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Demonstrates how to import core components from the aiogram_webhook package for setting up webhook functionality. ```python from aiogram_webhook import SingleBotEngine, FastAPIAdapter, Route from aiogram_webhook import BotConfig, WebhookConfig ``` -------------------------------- ### Install aiogram-webhook Source: https://github.com/m-xim/aiogram-webhook/blob/main/README.md Install the aiogram-webhook package. Additional dependencies for FastAPI or aiohttp can be installed using package extras. ```bash pip install aiogram-webhook ``` ```bash pip install "aiogram-webhook[fastapi]" ``` ```bash pip install "aiogram-webhook[aiohttp]" ``` -------------------------------- ### Single Static Query Parameter Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/query-parameters.md A complete example demonstrating a Route with a single static query parameter. Incoming requests must include this parameter with the specified value. ```python from aiogram_webhook import Route, Const route = Route( base_url="https://example.com", path="/webhook", query={"secret": "my_secret_key"} ) # URL: https://example.com/webhook?secret=my_secret_key # Incoming requests MUST have ?secret=my_secret_key ``` -------------------------------- ### Security Secret Token Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/security.md Example of retrieving and printing the secret token for a given bot target. ```python token = await security.secret_token(target) if token: print(f"Secret token for bot {target.bot_id}: {token}") ``` -------------------------------- ### SingleBotEngine Constructor Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/engines/overview.md Illustrates the typical arguments passed to the SingleBotEngine constructor. The 'security' and 'webhook_config' arguments are optional but recommended for production environments. ```python engine = SingleBotEngine( dispatcher, bot, web=adapter, route=route, security=security, # optional, recommended in production webhook_config=webhook_config, # optional Telegram setWebhook fields handle_in_background=True, # default; see Behavior ) ``` -------------------------------- ### on_startup Method Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/base-webhook-engine.md A lifecycle hook that is called when the web application starts. It is responsible for emitting the dispatcher's startup event. ```APIDOC ## on_startup Method ```python async def on_startup(self, app: AppT, *args: Any, **kwargs: Any) -> None: ``` ### Description Lifecycle hook called when the web application starts. Emits the dispatcher startup event. ### Parameters - `app` (AppT): The web application instance. - `*args`, `**kwargs`: Additional framework-specific arguments. ### Returns None ### Called by This method is invoked by the web adapter's lifecycle callbacks. ``` -------------------------------- ### Minimal TokenEngine Setup Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/engines/token-engine.md Sets up a basic TokenEngine with a Dispatcher, FastAPIAdapter, a dynamic route, and default bot/webhook configurations. Use this for a straightforward multi-bot webhook service. ```python from aiogram import Dispatcher from aiogram.client.default import DefaultBotProperties from aiogram.client.session.aiohttp import AiohttpSession from aiogram_webhook import BotConfig, FastAPIAdapter, TokenEngine, WebhookConfig from aiogram_webhook.route import BotTokenParam, Route dispatcher = Dispatcher() route = Route( base_url="https://example.com", path="/webhook/{bot_token}", params={"bot_token": BotTokenParam()}, ) engine = TokenEngine( dispatcher, web=FastAPIAdapter(), route=route, bot_config=BotConfig(default=DefaultBotProperties(parse_mode="HTML")), webhook_config=WebhookConfig(drop_pending_updates=True), ) ``` -------------------------------- ### IPCheck Examples Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/security.md Demonstrates different ways to configure the IPCheck, including allowing Telegram defaults, excluding them, or specifying custom IP networks and addresses. ```python from aiogram_webhook.security.checks.ip import IPCheck # Allow only Telegram and one additional IP check = IPCheck("192.168.1.100", "10.0.0.0/8", include_default=True) # Allow only specified networks, exclude Telegram defaults check = IPCheck("192.168.0.0/16", include_default=False) # Allow specific IP addresses and networks check = IPCheck( "203.0.113.50", "203.0.113.0/24", "2001:db8::/32" ) ``` -------------------------------- ### Implement DatabaseSecretToken Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/types.md Example of a SecretToken implementation that fetches the secret token from a database. Use this for per-bot secret token management. ```python class DatabaseSecretToken(SecretToken): async def secret_token(self, target: Target) -> str: # Fetch from database return await db.get_bot_secret_token(target.bot_id) # Use in Security security = Security(secret_token=DatabaseSecretToken()) ``` -------------------------------- ### TokenEngine Setup and Bot Management Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/README.md Initialize a TokenEngine for multi-bot applications where bot tokens are extracted from the URL. This engine dynamically manages multiple bot instances. ```python from aiogram_webhook import TokenEngine, BotTokenParam engine = TokenEngine( dispatcher=dispatcher, web=FastAPIAdapter(), route=Route( base_url="https://example.com", path="/webhook/{bot_token}", params={\"bot_token\": BotTokenParam()} ) ) # Dynamically add/remove bots bot = await engine.add_bot("123:ABC-token") await engine.remove_bot(123, delete_webhook=True) ``` -------------------------------- ### BotTokenParam.build Method Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/bot-token-param.md Demonstrates how to use the build method of BotTokenParam to convert a Target object into a string representation of the bot token for URL paths. The token is returned as-is. ```python from aiogram_webhook.engines.target import Target from aiogram_webhook.route.params import BotTokenParam target = Target(bot_id=123456789, bot_token="123456789:ABCDEFtoken123") param = BotTokenParam() token_str = await param.build(target, {}) # Result: "123456789:ABCDEFtoken123" ``` -------------------------------- ### TaskTracker Constructor Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/task-tracker.md Initializes a new TaskTracker instance. It does not take any parameters and starts with an empty set of tracked tasks. ```APIDOC ## TaskTracker Constructor Creates a new TaskTracker with no parameters. ```python class TaskTracker: def __init__(self) -> None: ``` ``` -------------------------------- ### SingleBotEngine Setup Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/README.md Initialize and register a SingleBotEngine for a single bot application. This engine handles webhook registration, request processing, and lifecycle management for one Telegram bot. ```python from aiogram_webhook import SingleBotEngine, FastAPIAdapter, Route engine = SingleBotEngine( dispatcher=dispatcher, bot=bot, web=FastAPIAdapter(), route=Route(base_url="https://example.com", path="/webhook") ) engine.register(app) await engine.set_webhook() ``` -------------------------------- ### URL Building with Query Parameters Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/query-parameters.md Shows how query parameters are incorporated when building webhook URLs using a defined route. This example demonstrates using both static and dynamic query parameters. ```python from aiogram_webhook import Route, BotIdParam, Const, Ref from aiogram_webhook.engines.target import Target route = Route( base_url="https://example.com", path="/webhook/{bot_id}", params={"bot_id": BotIdParam()}, query={ "secret": Const("key123"), "bot_id": Ref("bot_id") } ) target = Target(bot_id=987654321, bot_token="987654321:token") url = await route.build_url(target=target) # Result: "https://example.com/webhook/987654321?secret=key123&bot_id=987654321" ``` -------------------------------- ### Handle SecurityCheckError Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/errors.md Example of catching a SecurityCheckError and printing details about the failed security check and the client IP. ```python # Raised if IPCheck rejects client IP try: security.verify(target=target, request=request, route_params=route_params) except SecurityCheckError as e: print(f"Check failed: {e.security_check} from IP {e.client_ip}") ``` -------------------------------- ### Spawn Coroutine with TaskTracker Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/task-tracker.md Starts a coroutine in the background and registers it with the TaskTracker for lifecycle management. Returns the created asyncio Task object. ```python async def background_work(): await asyncio.sleep(1) print("Work complete") tracker = TaskTracker() task = tracker.spawn(background_work()) # Task runs in background while main code continues ``` -------------------------------- ### Add Bot to TokenEngine Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Adds and registers a new bot with the TokenEngine using its token. This is a common setup step for multi-bot configurations. ```python from aiogram_webhook import TokenEngine engine = TokenEngine(dispatcher=dispatcher, web=adapter, route=route) bot = await engine.add_bot("123:ABC-token") ``` -------------------------------- ### Simple Single-Bot Webhook Route Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/route.md Example of creating a basic Route for a single-bot webhook with a fixed base URL and path. No parameters are defined in this simple case. ```python from aiogram_webhook import Route route = Route( base_url="https://example.com", path="/webhook" ) # URL: https://example.com/webhook ``` -------------------------------- ### Token-Based Multi-Bot Setup with TokenEngine Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/bot-token-param.md Configures a Route to extract the full bot token from the URL path and uses TokenEngine to automatically create bots based on these tokens. Requests to paths like /webhook/123:ABC-token are routed to the corresponding bot. ```python from aiogram_webhook import Route, BotTokenParam, TokenEngine # Route extracts full bot token from path route = Route( base_url="https://api.example.com", path="/webhook/{bot_token}", params={"bot_token": BotTokenParam()} ) engine = TokenEngine( dispatcher=dispatcher, web=web_adapter, route=route ) # Requests to /webhook/123:ABC-token automatically route to bot 123 # TokenEngine creates bot with that token ``` -------------------------------- ### Request Matching with Query Specifications Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/query-parameters.md Explains how incoming requests are validated against defined query specifications. This snippet highlights valid and invalid request examples and the errors that can be raised. ```python from aiogram_webhook import Route, Const route = Route( base_url="https://example.com", path="/webhook", query={"key": "value"} ) # Valid request: POST /webhook?key=value # Invalid request: POST /webhook (no query) # Invalid request: POST /webhook?key=wrong # Invalid request: POST /webhook?different=param # route.match(request) raises MissingQueryParamError or QueryParamMismatchError ``` -------------------------------- ### Complex Route with All Features Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/query-parameters.md Illustrates a multi-bot route configuration with comprehensive validation for path and query parameters, including dynamic parameters and fixed constants. This setup supports multiple bots and various data formats. ```python from aiogram_webhook import Route, BotIdParam, BotTokenParam, Const, Ref # Multi-bot route with comprehensive validation route = Route( base_url="https://api.example.com:8443", path="/bots/{bot_id}/webhook/{bot_token}", params={ "bot_id": BotIdParam(), "bot_token": BotTokenParam() }, query={ "api_version": Const("v7"), "client": Const("telegram"), "bot_id": Ref("bot_id"), "formats": ["json", "protobuf"] }, strict_query=True ) # Generated URL: # https://api.example.com:8443/bots/123456789/webhook/123456789:ABCtoken/ # ?api_version=v7&client=telegram&bot_id=123456789&formats=json&formats=protobuf ``` -------------------------------- ### Strict Query Validation Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/query-parameters.md Demonstrates how to configure a route to strictly validate query parameters, rejecting requests with unexpected parameters. An UnexpectedQueryParamError is raised with a 404 status if validation fails. ```python from aiogram_webhook import Route route = Route( base_url="https://example.com", path="/webhook", query={"secret": "key123"}, strict_query=True ) # Incoming request: /webhook?secret=key123 ✓ OK # Incoming request: /webhook?secret=key123&extra=value ✗ REJECTED # UnexpectedQueryParamError raised with status 404 ``` -------------------------------- ### BotTokenParam.parse Method Example Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/bot-token-param.md Shows how to use the parse method of BotTokenParam to extract a bot token string from a URL path value. Invalid tokens will raise a ValueError, which is typically handled by the engine. ```python param = BotTokenParam() # Valid token token = await param.parse("123456789:ABCDEFtoken123", {}) # Result: "123456789:ABCDEFtoken123" # Invalid token will be caught by engine and raise TokenValidationError ``` -------------------------------- ### Production Security Setup with IP Whitelisting Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/README.md Set up security with specific Telegram IP ranges and a static secret token from an environment variable. Ensures requests originate from trusted IPs and possess the correct token. ```python from aiogram_webhook.security import Security, StaticSecretToken from aiogram_webhook.security.checks.ip import IPCheck security = Security( IPCheck( "149.154.160.0/20", # Telegram IP range 1 "91.108.4.0/22", # Telegram IP range 2 include_default=False # Only allow explicit ranges ), secret_token=StaticSecretToken(os.getenv("WEBHOOK_SECRET")) ) ``` -------------------------------- ### _on_startup Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/base-webhook-engine.md Abstract method for engine-specific startup logic. It emits a dispatcher startup event with lifecycle data and must be implemented by subclasses. ```APIDOC ## _on_startup ### Description Engine-specific startup logic. Emits dispatcher startup event with lifecycle data. ### Parameters - `app` (AppT) - Required - The web application. - `*args`, `**kwargs` - Optional - Framework-specific arguments. ### Implemented by - SingleBotEngine: Emits startup with single bot - TokenEngine: Emits startup with all managed bots ``` -------------------------------- ### ID-Based Multi-Bot Setup with Custom Engine Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/bot-token-param.md Sets up a Route to extract a numeric bot ID from the URL path. A custom BaseMultiBotEngine subclass is defined to resolve the bot target by looking up the bot token from a database using the extracted ID. ```python from aiogram_webhook import Route, BotIdParam from aiogram_webhook.engines.multi import BaseMultiBotEngine from aiogram_webhook.route.params import RouteParams from aiogram_webhook.engines.target import Target # Route extracts numeric bot ID route = Route( base_url="https://api.example.com", path="/webhook/{bot_id}", params={"bot_id": BotIdParam()} ) # Custom engine looks up bot by ID from database class DatabaseMultiBotEngine(BaseMultiBotEngine): async def _resolve_target(self, request, route_params): bot_id = route_params.get("bot_id") # Lookup bot token from database bot_token = await db.get_bot_token(bot_id) if bot_token: return Target(bot_id=bot_id, bot_token=bot_token) return None ``` -------------------------------- ### _on_startup Abstract Method Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/base-webhook-engine.md Abstract method for engine-specific startup logic. Emits a dispatcher startup event with lifecycle data. Subclasses must implement this to perform actions upon engine startup. ```python @abstractmethod async def _on_startup(self, app: AppT, *args: Any, **kwargs: Any) -> None: # ... implementation details ... ``` -------------------------------- ### Handle InvalidBaseUrlError Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/errors.md Example of catching an InvalidBaseUrlError when creating a Route with an invalid base_url. ```python try: route = Route(base_url="invalid-url", path="/webhook") except InvalidBaseUrlError as e: print(f"Invalid base_url {e.base_url}: {e.reason}") ``` -------------------------------- ### Handle MissingRouteParamDeclarationError Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/errors.md Example of catching a MissingRouteParamDeclarationError when a Route is configured with a path parameter that is not declared in the `params` dictionary. ```python try: route = Route( base_url="https://example.com", path="/webhook/{bot_id}", params={} # Missing bot_id declaration ) except MissingRouteParamDeclarationError as e: print(f"Missing params: {e.missing_params}") ``` -------------------------------- ### Handle RepeatedPathParamError Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/errors.md Example of catching a RepeatedPathParamError when a Route is defined with a path containing duplicate parameter names. ```python try: route = Route( base_url="https://example.com", path="/webhook/{id}/{id}" # Repeated parameter ) except RepeatedPathParamError as e: print(f"Repeated params: {e.repeated_params}") ``` -------------------------------- ### Get Path Parameters Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/web-adapters.md Retrieve path parameters from the route template. This method is part of the WebAdapter interface. ```python @property def path_params(self) -> PathParams: ``` -------------------------------- ### Configure SingleBotEngine Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/configuration.md Instantiate SingleBotEngine with dispatcher, bot, web adapter, route, and optional security and background processing settings. ```python from aiogram_webhook import SingleBotEngine, FastAPIAdapter, Route engine = SingleBotEngine( dispatcher=dispatcher, bot=bot, web=FastAPIAdapter(), route=Route(base_url="https://example.com", path="/webhook"), security=None, # Add security=Security(...) for production handle_in_background=True, shutdown_timeout=10.0 ) ``` -------------------------------- ### TokenEngine Initialization Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Initializes the TokenEngine for managing multiple bots using token-based routing. Requires a dispatcher, web adapter, and route configuration. ```python class TokenEngine(BaseMultiBotEngine[AppT, RawRequestT, FrameworkResponseT]): def __init__( self, dispatcher, web: WebAdapter[AppT, RawRequestT, FrameworkResponseT], route: Route, security=None, bot_config: BotConfig | None = None, webhook_config: WebhookConfig | None = None, handle_in_background: bool = True, shutdown_timeout: float = 10.0, ) -> None: ``` -------------------------------- ### Route Definition for Bot ID Engine Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/engines/custom-engine.md Example of how to define a route that includes a 'bot_id' parameter, which can be used by a custom engine. ```python from aiogram_webhook.route import BotIdParam, Route Route( base_url="https://example.com", path="/webhook/{bot_id}", params={"bot_id": BotIdParam()}, ) ``` -------------------------------- ### _get_task_tracker Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/base-webhook-engine.md Abstract method to get the TaskTracker for a bot instance. This is used for spawning background tasks and must be implemented by subclasses. ```APIDOC ## _get_task_tracker ### Description Get the TaskTracker for a bot instance. Used for spawning background tasks. ### Parameters - `bot` (Bot) - Required - The bot instance. ### Returns `TaskTracker` — Task tracker for the bot. ### Implemented by - SingleBotEngine: Returns single shared TaskTracker - TokenEngine: Returns per-bot TaskTracker from cache ``` -------------------------------- ### Recommended Main Imports Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Commonly used classes for setting up aiogram-webhook. ```python # Main classes from aiogram_webhook import ( SingleBotEngine, TokenEngine, FastAPIAdapter, AiohttpAdapter, Route, BotConfig, WebhookConfig, ) ``` -------------------------------- ### Register Bots at Startup Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/recipes/multi-bot.md This asynchronous function demonstrates how to register multiple bots with the engine during application startup. Each bot is added using its unique token. ```python async def on_startup(app): await engine.add_bot("123456:ABCDEF") await engine.add_bot("654321:UVWXYZ") ``` -------------------------------- ### spawn Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/task-tracker.md Starts a coroutine in the background and adds it to the tracker for lifecycle management. It returns the asyncio.Task object representing the spawned coroutine. ```APIDOC ## spawn Start a coroutine in the background and track it for lifecycle management. ### Method `spawn` ### Parameters - `coro` (Coroutine): The coroutine to execute ### Returns `asyncio.Task[TaskResultT]` — The created asyncio Task object ### Example ```python async def background_work(): await asyncio.sleep(1) print("Work complete") tracker = TaskTracker() task = tracker.spawn(background_work()) # Task runs in background while main code continues ``` ``` -------------------------------- ### Top-Level Package Exports Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Imports commonly used classes from the main aiogram_webhook package. Conditionally exports FastAPIAdapter if fastapi is installed. ```python from aiogram_webhook import ( AiohttpAdapter, BotConfig, SingleBotEngine, TokenEngine, WebhookConfig, ) # Conditionally exported (if fastapi installed): # from aiogram_webhook import FastAPIAdapter ``` -------------------------------- ### Implement Custom SecurityCheck Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/types.md Example of a custom SecurityCheck that verifies a custom header. Use this to implement specific security rules for your webhook. ```python class CustomSecurityCheck(SecurityCheck): async def verify(self, target: Target, request: WebRequest, route_params: RouteParams) -> bool: # Custom verification logic return request.headers.get("X-Custom-Header") == "expected_value" # Use in Security security = Security( CustomSecurityCheck(), secret_token=StaticSecretToken("token123") ) ``` -------------------------------- ### BaseWebhookEngine Constructor Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/base-webhook-engine.md Initializes the BaseWebhookEngine with essential components for handling webhook requests. It requires a Dispatcher, WebAdapter, and Route, with optional Security configuration and background processing settings. ```APIDOC ## BaseWebhookEngine Constructor ```python class BaseWebhookEngine(ABC, Generic[AppT, RawRequestT, FrameworkResponseT]): def __init__( self, dispatcher: Dispatcher, web: WebAdapter[AppT, RawRequestT, FrameworkResponseT], route: Route, security: Security | None = None, handle_in_background: bool = True, shutdown_timeout: float = 10.0, ) -> None: ``` ### Parameters | Parameter | Type | Required | Default | Description | |-----------|------|----------|---------|-------------| | `dispatcher` | `Dispatcher` | Yes | — | aiogram Dispatcher instance | | `web` | `WebAdapter` | Yes | — | Web framework adapter | | `route` | `Route` | Yes | — | Route configuration | | `security` | `Security | None` | No | `None` | Security configuration (warning if None) | | `handle_in_background` | `bool` | No | `True` | Process updates asynchronously | | `shutdown_timeout` | `float` | No | `10.0` | Background task shutdown timeout | ``` -------------------------------- ### Handle RequestHandlingStoppedError Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/errors.md Example of catching `RequestHandlingStoppedError`. This error is raised when a request arrives during the engine's shutdown process, preventing further handling. ```python # Raised if request arrives during on_shutdown phase try: engine.handle_request(request) except RequestHandlingStoppedError: print("Engine is shutting down, request rejected") ``` -------------------------------- ### Configure Security with IP Check and Secret Token Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/configuration.md Instantiate Security with IP address checks and a static secret token for webhook verification. ```python from aiogram_webhook.security import Security, StaticSecretToken from aiogram_webhook.security.checks.ip import IPCheck security = Security( IPCheck(include_default=True), secret_token=StaticSecretToken("my_secret_token_123") ) ``` -------------------------------- ### Minimal Custom WebAdapter Skeleton Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/web/custom.md A basic implementation of a custom WebAdapter. Replace framework-specific methods like `app.post` and lifecycle hooks with your framework's equivalents. ```python from aiogram_webhook.web.base import WebAdapter class MyAdapter(WebAdapter): def bind_request(self, request): return MyWebRequest(request) def register(self, app, path, handler, *, on_startup, on_shutdown) -> None: async def endpoint(raw_request): return await handler(self.bind_request(raw_request)) app.post(path, endpoint) app.on_startup(on_startup) app.on_shutdown(on_shutdown) def json_response(self, status_code: int, data=None, headers=None): return my_json_response(status_code=status_code, data=data, headers=headers) def payload_response(self, status_code: int, payload, headers=None): return my_payload_response(status_code=status_code, payload=payload, headers=headers) ``` -------------------------------- ### Implementing Custom Query Values with QueryValue Protocol Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/query-parameters.md Demonstrates how to create custom query parameter handlers by implementing the QueryValue protocol. This allows for advanced scenarios like hashing parameter values for checksums. ```python from aiogram_webhook.route.query import QueryValue from aiogram_webhook.route.params import RouteParams class HashRef(QueryValue): """Query value that is a hash of a route parameter.""" def __init__(self, param_name: str): self.param_name = param_name def render(self, params: RouteParams) -> str: import hashlib value = str(params[self.param_name]) return hashlib.sha256(value.encode()).hexdigest() def required_params(self) -> frozenset[str]: return frozenset((self.param_name,)) # Use in route route = Route( base_url="https://example.com", path="/webhook/{bot_id}", params={"bot_id": BotIdParam()}, query={"checksum": HashRef("bot_id")} ) ``` -------------------------------- ### TokenEngine Constructor Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/token-engine.md Initializes the TokenEngine with necessary components for managing multiple bots. It requires a dispatcher, a web adapter, and a route configuration that includes a bot token parameter. ```APIDOC ## TokenEngine Constructor ### Description Initializes the TokenEngine with necessary components for managing multiple bots. It requires a dispatcher, a web adapter, and a route configuration that includes a bot token parameter. ### Parameters - **dispatcher** (`Dispatcher`) - Required - aiogram Dispatcher instance for update processing - **web** (`WebAdapter[AppT, RawRequestT, FrameworkResponseT]`) - Required - Web framework adapter (FastAPIAdapter or AiohttpAdapter) - **route** (`Route`) - Required - Route configuration defining webhook path with `{bot_token}` parameter - **security** (`Security | None`) - Optional - Security configuration for verifying incoming requests - **bot_config** (`BotConfig | None`) - Optional - Configuration for bot creation (session, default properties) - **webhook_config** (`WebhookConfig | None`) - Optional - Default webhook configuration for setWebhook calls - **handle_in_background** (`bool`) - Optional - Default: `True`. If True, updates are processed as background tasks - **shutdown_timeout** (`float`) - Optional - Default: `10.0`. Timeout in seconds for gracefully closing tasks on shutdown ``` -------------------------------- ### Static Secret Token Validation Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/MODULES.md Use StaticSecretToken to validate incoming requests against a predefined secret token. This is useful for simple security setups. ```python from aiogram_webhook.security.secret_token import StaticSecretToken secret_token = StaticSecretToken("your-secret-token") ``` -------------------------------- ### TokenEngine Constructor Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/token-engine.md Initializes the TokenEngine with essential components like a dispatcher, web adapter, and route configuration. Optional parameters allow for security, bot, and webhook configurations, as well as background task handling and shutdown timeouts. ```python class TokenEngine(BaseMultiBotEngine[AppT, RawRequestT, FrameworkResponseT]): def __init__( self, dispatcher: Dispatcher, web: WebAdapter[AppT, RawRequestT, FrameworkResponseT], route: Route, security: Security | None = None, bot_config: BotConfig | None = None, webhook_config: WebhookConfig | None = None, handle_in_background: bool = True, shutdown_timeout: float = 10.0, ) -> None: ``` -------------------------------- ### _resolve_bot Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/api-reference/base-webhook-engine.md Abstract method to get or create the Bot instance for a given target. This method is called during request handling and must be implemented by subclasses. ```APIDOC ## _resolve_bot ### Description Get or create the Bot instance for a given target. Called during request handling. ### Parameters - `target` (Target) - Required - The target bot identifier. ### Returns `Bot | None` — The Bot instance, or None if bot cannot be resolved. ### Implemented by - SingleBotEngine: Returns self.bot - TokenEngine: Creates or retrieves bot from internal cache ``` -------------------------------- ### Get Webhook Info Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/learn/first-webhook.md Retrieve information about the currently registered webhook, including its URL and the last error message if any. This helps in debugging connectivity issues. ```python info = await bot.get_webhook_info() print(info.url, info.last_error_message) ``` -------------------------------- ### Handle InvalidPathParamError Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/errors.md Example of catching an InvalidPathParamError when a route parameter fails to parse its value from the incoming request, such as trying to parse non-integer input for an integer parameter. ```python # If BotIdParam tries to parse "abc" as integer try: route_params = await route.match(request) except InvalidPathParamError as e: print(f"Cannot parse {e.param}='{e.value}'") ``` -------------------------------- ### Basic Route Configuration Source: https://github.com/m-xim/aiogram-webhook/blob/main/docs/route/overview.md Defines a route with a base URL, a path containing a bot token parameter, and query parameters referencing the bot token and a constant value. ```python from aiogram_webhook.route import BotTokenParam, Ref, Route route = Route( base_url="https://example.com/api", path="/telegram/{bot_token}", params={\"bot_token\": BotTokenParam()}, query={\"token\": Ref(\"bot_token\"), \"kind\": (\"telegram\", \"webhook\")}, ) ``` -------------------------------- ### Handle SecretTokenError Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/errors.md Example of catching `SecretTokenError`, which occurs when the provided secret token for a bot does not match the expected value. It logs the bot ID for which the token was invalid. ```python # Raised if secret token doesn't match expected value from aiogram_webhook.security.errors import SecretTokenError try: security.verify(target=target, request=request, route_params=route_params) except SecretTokenError as e: print(f"Invalid secret token for bot {e.target_bot_id}") ``` -------------------------------- ### Implement Custom RouteParam Source: https://github.com/m-xim/aiogram-webhook/blob/main/_autodocs/types.md Example of a custom RouteParam implementation that converts parsed values to uppercase. Use this to define custom parsing logic for route parameters. ```python class CustomParam(RouteParam): async def build(self, target: Target, params: RouteParams) -> str: return "custom_value" async def parse(self, value: str, params: RouteParams) -> str: return value.upper() # Use in Route route = Route( base_url="https://example.com", path="/webhook/{custom}", params={"custom": CustomParam()} ) ```