customwebhookbot.py

This example is available for different web frameworks. You can select your preferred framework by opening one of the tabs above the code example.

Hint

The following examples show how different Python web frameworks can be used alongside PTB. This can be useful for two use cases:

  1. For extending the functionality of your existing bot to handling updates of external services

  2. For extending the functionality of your exisiting web application to also include chat bot functionality

How the PTB and web framework components of the examples below are viewed surely depends on which use case one has in mind. We are fully aware that a combination of PTB with web frameworks will always mean finding a tradeoff between usability and best practices for both PTB and the web framework and these examples are certainly far from optimal solutions. Please understand them as starting points and use your expertise of the web framework of your choosing to build up on them. You are of course also very welcome to help improve these examples!

  1#!/usr/bin/env python
  2# This program is dedicated to the public domain under the CC0 license.
  3# pylint: disable=import-error,unused-argument
  4"""
  5Simple example of a bot that uses a custom webhook setup and handles custom updates.
  6For the custom webhook setup, the libraries `starlette` and `uvicorn` are used. Please install
  7them as `pip install starlette~=0.20.0 uvicorn~=0.23.2`.
  8Note that any other `asyncio` based web server framework can be used for a custom webhook setup
  9just as well.
 10
 11Usage:
 12Set bot Token, URL, admin CHAT_ID and PORT after the imports.
 13You may also need to change the `listen` value in the uvicorn configuration to match your setup.
 14Press Ctrl-C on the command line or send a signal to the process to stop the bot.
 15"""
 16
 17import asyncio
 18import html
 19import logging
 20from dataclasses import dataclass
 21from http import HTTPStatus
 22
 23import uvicorn
 24from starlette.applications import Starlette
 25from starlette.requests import Request
 26from starlette.responses import PlainTextResponse, Response
 27from starlette.routing import Route
 28
 29from telegram import Update
 30from telegram.constants import ParseMode
 31from telegram.ext import (
 32    Application,
 33    CallbackContext,
 34    CommandHandler,
 35    ContextTypes,
 36    ExtBot,
 37    TypeHandler,
 38)
 39
 40# Enable logging
 41logging.basicConfig(
 42    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
 43)
 44# set higher logging level for httpx to avoid all GET and POST requests being logged
 45logging.getLogger("httpx").setLevel(logging.WARNING)
 46
 47logger = logging.getLogger(__name__)
 48
 49# Define configuration constants
 50URL = "https://domain.tld"
 51ADMIN_CHAT_ID = 123456
 52PORT = 8000
 53TOKEN = "123:ABC"  # nosec B105
 54
 55
 56@dataclass
 57class WebhookUpdate:
 58    """Simple dataclass to wrap a custom update type"""
 59
 60    user_id: int
 61    payload: str
 62
 63
 64class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
 65    """
 66    Custom CallbackContext class that makes `user_data` available for updates of type
 67    `WebhookUpdate`.
 68    """
 69
 70    @classmethod
 71    def from_update(
 72        cls,
 73        update: object,
 74        application: "Application",
 75    ) -> "CustomContext":
 76        if isinstance(update, WebhookUpdate):
 77            return cls(application=application, user_id=update.user_id)
 78        return super().from_update(update, application)
 79
 80
 81async def start(update: Update, context: CustomContext) -> None:
 82    """Display a message with instructions on how to use this bot."""
 83    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
 84    text = (
 85        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
 86        f"To post a custom update, call <code>{payload_url}</code>."
 87    )
 88    await update.message.reply_html(text=text)
 89
 90
 91async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
 92    """Handle custom updates."""
 93    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
 94    payloads = context.user_data.setdefault("payloads", [])
 95    payloads.append(update.payload)
 96    combined_payloads = "</code>\n• <code>".join(payloads)
 97    text = (
 98        f"The user {chat_member.user.mention_html()} has sent a new payload. "
 99        f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
100    )
101    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
102
103
104async def main() -> None:
105    """Set up PTB application and a web application for handling the incoming requests."""
106    context_types = ContextTypes(context=CustomContext)
107    # Here we set updater to None because we want our custom webhook server to handle the updates
108    # and hence we don't need an Updater instance
109    application = (
110        Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
111    )
112
113    # register handlers
114    application.add_handler(CommandHandler("start", start))
115    application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
116
117    # Pass webhook settings to telegram
118    await application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
119
120    # Set up webserver
121    async def telegram(request: Request) -> Response:
122        """Handle incoming Telegram updates by putting them into the `update_queue`"""
123        await application.update_queue.put(
124            Update.de_json(data=await request.json(), bot=application.bot)
125        )
126        return Response()
127
128    async def custom_updates(request: Request) -> PlainTextResponse:
129        """
130        Handle incoming webhook updates by also putting them into the `update_queue` if
131        the required parameters were passed correctly.
132        """
133        try:
134            user_id = int(request.query_params["user_id"])
135            payload = request.query_params["payload"]
136        except KeyError:
137            return PlainTextResponse(
138                status_code=HTTPStatus.BAD_REQUEST,
139                content="Please pass both `user_id` and `payload` as query parameters.",
140            )
141        except ValueError:
142            return PlainTextResponse(
143                status_code=HTTPStatus.BAD_REQUEST,
144                content="The `user_id` must be a string!",
145            )
146
147        await application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
148        return PlainTextResponse("Thank you for the submission! It's being forwarded.")
149
150    async def health(_: Request) -> PlainTextResponse:
151        """For the health endpoint, reply with a simple plain text message."""
152        return PlainTextResponse(content="The bot is still running fine :)")
153
154    starlette_app = Starlette(
155        routes=[
156            Route("/telegram", telegram, methods=["POST"]),
157            Route("/healthcheck", health, methods=["GET"]),
158            Route("/submitpayload", custom_updates, methods=["POST", "GET"]),
159        ]
160    )
161    webserver = uvicorn.Server(
162        config=uvicorn.Config(
163            app=starlette_app,
164            port=PORT,
165            use_colors=False,
166            host="127.0.0.1",
167        )
168    )
169
170    # Run application and webserver together
171    async with application:
172        await application.start()
173        await webserver.serve()
174        await application.stop()
175
176
177if __name__ == "__main__":
178    asyncio.run(main())

  1#!/usr/bin/env python
  2# This program is dedicated to the public domain under the CC0 license.
  3# pylint: disable=import-error,unused-argument
  4"""
  5Simple example of a bot that uses a custom webhook setup and handles custom updates.
  6For the custom webhook setup, the libraries `flask`, `asgiref` and `uvicorn` are used. Please
  7install them as `pip install flask[async]~=2.3.2 uvicorn~=0.23.2 asgiref~=3.7.2`.
  8Note that any other `asyncio` based web server framework can be used for a custom webhook setup
  9just as well.
 10
 11Usage:
 12Set bot Token, URL, admin CHAT_ID and PORT after the imports.
 13You may also need to change the `listen` value in the uvicorn configuration to match your setup.
 14Press Ctrl-C on the command line or send a signal to the process to stop the bot.
 15"""
 16
 17import asyncio
 18import html
 19import logging
 20from dataclasses import dataclass
 21from http import HTTPStatus
 22
 23import uvicorn
 24from asgiref.wsgi import WsgiToAsgi
 25from flask import Flask, Response, abort, make_response, request
 26
 27from telegram import Update
 28from telegram.constants import ParseMode
 29from telegram.ext import (
 30    Application,
 31    CallbackContext,
 32    CommandHandler,
 33    ContextTypes,
 34    ExtBot,
 35    TypeHandler,
 36)
 37
 38# Enable logging
 39logging.basicConfig(
 40    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
 41)
 42# set higher logging level for httpx to avoid all GET and POST requests being logged
 43logging.getLogger("httpx").setLevel(logging.WARNING)
 44
 45logger = logging.getLogger(__name__)
 46
 47# Define configuration constants
 48URL = "https://domain.tld"
 49ADMIN_CHAT_ID = 123456
 50PORT = 8000
 51TOKEN = "123:ABC"  # nosec B105
 52
 53
 54@dataclass
 55class WebhookUpdate:
 56    """Simple dataclass to wrap a custom update type"""
 57
 58    user_id: int
 59    payload: str
 60
 61
 62class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
 63    """
 64    Custom CallbackContext class that makes `user_data` available for updates of type
 65    `WebhookUpdate`.
 66    """
 67
 68    @classmethod
 69    def from_update(
 70        cls,
 71        update: object,
 72        application: "Application",
 73    ) -> "CustomContext":
 74        if isinstance(update, WebhookUpdate):
 75            return cls(application=application, user_id=update.user_id)
 76        return super().from_update(update, application)
 77
 78
 79async def start(update: Update, context: CustomContext) -> None:
 80    """Display a message with instructions on how to use this bot."""
 81    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
 82    text = (
 83        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
 84        f"To post a custom update, call <code>{payload_url}</code>."
 85    )
 86    await update.message.reply_html(text=text)
 87
 88
 89async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
 90    """Handle custom updates."""
 91    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
 92    payloads = context.user_data.setdefault("payloads", [])
 93    payloads.append(update.payload)
 94    combined_payloads = "</code>\n• <code>".join(payloads)
 95    text = (
 96        f"The user {chat_member.user.mention_html()} has sent a new payload. "
 97        f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
 98    )
 99    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
100
101
102async def main() -> None:
103    """Set up PTB application and a web application for handling the incoming requests."""
104    context_types = ContextTypes(context=CustomContext)
105    # Here we set updater to None because we want our custom webhook server to handle the updates
106    # and hence we don't need an Updater instance
107    application = (
108        Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
109    )
110
111    # register handlers
112    application.add_handler(CommandHandler("start", start))
113    application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
114
115    # Pass webhook settings to telegram
116    await application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
117
118    # Set up webserver
119    flask_app = Flask(__name__)
120
121    @flask_app.post("/telegram")  # type: ignore[misc]
122    async def telegram() -> Response:
123        """Handle incoming Telegram updates by putting them into the `update_queue`"""
124        await application.update_queue.put(Update.de_json(data=request.json, bot=application.bot))
125        return Response(status=HTTPStatus.OK)
126
127    @flask_app.route("/submitpayload", methods=["GET", "POST"])  # type: ignore[misc]
128    async def custom_updates() -> Response:
129        """
130        Handle incoming webhook updates by also putting them into the `update_queue` if
131        the required parameters were passed correctly.
132        """
133        try:
134            user_id = int(request.args["user_id"])
135            payload = request.args["payload"]
136        except KeyError:
137            abort(
138                HTTPStatus.BAD_REQUEST,
139                "Please pass both `user_id` and `payload` as query parameters.",
140            )
141        except ValueError:
142            abort(HTTPStatus.BAD_REQUEST, "The `user_id` must be a string!")
143
144        await application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
145        return Response(status=HTTPStatus.OK)
146
147    @flask_app.get("/healthcheck")  # type: ignore[misc]
148    async def health() -> Response:
149        """For the health endpoint, reply with a simple plain text message."""
150        response = make_response("The bot is still running fine :)", HTTPStatus.OK)
151        response.mimetype = "text/plain"
152        return response
153
154    webserver = uvicorn.Server(
155        config=uvicorn.Config(
156            app=WsgiToAsgi(flask_app),
157            port=PORT,
158            use_colors=False,
159            host="127.0.0.1",
160        )
161    )
162
163    # Run application and webserver together
164    async with application:
165        await application.start()
166        await webserver.serve()
167        await application.stop()
168
169
170if __name__ == "__main__":
171    asyncio.run(main())

  1#!/usr/bin/env python
  2# This program is dedicated to the public domain under the CC0 license.
  3# pylint: disable=import-error,unused-argument
  4"""
  5Simple example of a bot that uses a custom webhook setup and handles custom updates.
  6For the custom webhook setup, the libraries `quart` and `uvicorn` are used. Please
  7install them as `pip install quart~=0.18.4 uvicorn~=0.23.2`.
  8Note that any other `asyncio` based web server framework can be used for a custom webhook setup
  9just as well.
 10
 11Usage:
 12Set bot Token, URL, admin CHAT_ID and PORT after the imports.
 13You may also need to change the `listen` value in the uvicorn configuration to match your setup.
 14Press Ctrl-C on the command line or send a signal to the process to stop the bot.
 15"""
 16
 17import asyncio
 18import html
 19import logging
 20from dataclasses import dataclass
 21from http import HTTPStatus
 22
 23import uvicorn
 24from quart import Quart, Response, abort, make_response, request
 25
 26from telegram import Update
 27from telegram.constants import ParseMode
 28from telegram.ext import (
 29    Application,
 30    CallbackContext,
 31    CommandHandler,
 32    ContextTypes,
 33    ExtBot,
 34    TypeHandler,
 35)
 36
 37# Enable logging
 38logging.basicConfig(
 39    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
 40)
 41# set higher logging level for httpx to avoid all GET and POST requests being logged
 42logging.getLogger("httpx").setLevel(logging.WARNING)
 43
 44logger = logging.getLogger(__name__)
 45
 46# Define configuration constants
 47URL = "https://domain.tld"
 48ADMIN_CHAT_ID = 123456
 49PORT = 8000
 50TOKEN = "123:ABC"  # nosec B105
 51
 52
 53@dataclass
 54class WebhookUpdate:
 55    """Simple dataclass to wrap a custom update type"""
 56
 57    user_id: int
 58    payload: str
 59
 60
 61class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
 62    """
 63    Custom CallbackContext class that makes `user_data` available for updates of type
 64    `WebhookUpdate`.
 65    """
 66
 67    @classmethod
 68    def from_update(
 69        cls,
 70        update: object,
 71        application: "Application",
 72    ) -> "CustomContext":
 73        if isinstance(update, WebhookUpdate):
 74            return cls(application=application, user_id=update.user_id)
 75        return super().from_update(update, application)
 76
 77
 78async def start(update: Update, context: CustomContext) -> None:
 79    """Display a message with instructions on how to use this bot."""
 80    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
 81    text = (
 82        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
 83        f"To post a custom update, call <code>{payload_url}</code>."
 84    )
 85    await update.message.reply_html(text=text)
 86
 87
 88async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
 89    """Handle custom updates."""
 90    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
 91    payloads = context.user_data.setdefault("payloads", [])
 92    payloads.append(update.payload)
 93    combined_payloads = "</code>\n• <code>".join(payloads)
 94    text = (
 95        f"The user {chat_member.user.mention_html()} has sent a new payload. "
 96        f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
 97    )
 98    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
 99
100
101async def main() -> None:
102    """Set up PTB application and a web application for handling the incoming requests."""
103    context_types = ContextTypes(context=CustomContext)
104    # Here we set updater to None because we want our custom webhook server to handle the updates
105    # and hence we don't need an Updater instance
106    application = (
107        Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
108    )
109
110    # register handlers
111    application.add_handler(CommandHandler("start", start))
112    application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
113
114    # Pass webhook settings to telegram
115    await application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
116
117    # Set up webserver
118    quart_app = Quart(__name__)
119
120    @quart_app.post("/telegram")  # type: ignore[misc]
121    async def telegram() -> Response:
122        """Handle incoming Telegram updates by putting them into the `update_queue`"""
123        await application.update_queue.put(
124            Update.de_json(data=await request.get_json(), bot=application.bot)
125        )
126        return Response(status=HTTPStatus.OK)
127
128    @quart_app.route("/submitpayload", methods=["GET", "POST"])  # type: ignore[misc]
129    async def custom_updates() -> Response:
130        """
131        Handle incoming webhook updates by also putting them into the `update_queue` if
132        the required parameters were passed correctly.
133        """
134        try:
135            user_id = int(request.args["user_id"])
136            payload = request.args["payload"]
137        except KeyError:
138            abort(
139                HTTPStatus.BAD_REQUEST,
140                "Please pass both `user_id` and `payload` as query parameters.",
141            )
142        except ValueError:
143            abort(HTTPStatus.BAD_REQUEST, "The `user_id` must be a string!")
144
145        await application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
146        return Response(status=HTTPStatus.OK)
147
148    @quart_app.get("/healthcheck")  # type: ignore[misc]
149    async def health() -> Response:
150        """For the health endpoint, reply with a simple plain text message."""
151        response = await make_response("The bot is still running fine :)", HTTPStatus.OK)
152        response.mimetype = "text/plain"
153        return response
154
155    webserver = uvicorn.Server(
156        config=uvicorn.Config(
157            app=quart_app,
158            port=PORT,
159            use_colors=False,
160            host="127.0.0.1",
161        )
162    )
163
164    # Run application and webserver together
165    async with application:
166        await application.start()
167        await webserver.serve()
168        await application.stop()
169
170
171if __name__ == "__main__":
172    asyncio.run(main())

  1#!/usr/bin/env python
  2# This program is dedicated to the public domain under the CC0 license.
  3# pylint: disable=import-error,unused-argument
  4"""
  5Simple example of a bot that uses a custom webhook setup and handles custom updates.
  6For the custom webhook setup, the libraries `Django` and `uvicorn` are used. Please
  7install them as `pip install Django~=4.2.4 uvicorn~=0.23.2`.
  8Note that any other `asyncio` based web server framework can be used for a custom webhook setup
  9just as well.
 10
 11Usage:
 12Set bot Token, URL, admin CHAT_ID and PORT after the imports.
 13You may also need to change the `listen` value in the uvicorn configuration to match your setup.
 14Press Ctrl-C on the command line or send a signal to the process to stop the bot.
 15"""
 16
 17import asyncio
 18import html
 19import json
 20import logging
 21from dataclasses import dataclass
 22from uuid import uuid4
 23
 24import uvicorn
 25from django.conf import settings
 26from django.core.asgi import get_asgi_application
 27from django.http import HttpRequest, HttpResponse, HttpResponseBadRequest
 28from django.urls import path
 29
 30from telegram import Update
 31from telegram.constants import ParseMode
 32from telegram.ext import (
 33    Application,
 34    CallbackContext,
 35    CommandHandler,
 36    ContextTypes,
 37    ExtBot,
 38    TypeHandler,
 39)
 40
 41# Enable logging
 42logging.basicConfig(
 43    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
 44)
 45# set higher logging level for httpx to avoid all GET and POST requests being logged
 46logging.getLogger("httpx").setLevel(logging.WARNING)
 47
 48logger = logging.getLogger(__name__)
 49
 50# Define configuration constants
 51URL = "https://domain.tld"
 52ADMIN_CHAT_ID = 123456
 53PORT = 8000
 54TOKEN = "123:ABC"  # nosec B105
 55
 56
 57@dataclass
 58class WebhookUpdate:
 59    """Simple dataclass to wrap a custom update type"""
 60
 61    user_id: int
 62    payload: str
 63
 64
 65class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
 66    """
 67    Custom CallbackContext class that makes `user_data` available for updates of type
 68    `WebhookUpdate`.
 69    """
 70
 71    @classmethod
 72    def from_update(
 73        cls,
 74        update: object,
 75        application: "Application",
 76    ) -> "CustomContext":
 77        if isinstance(update, WebhookUpdate):
 78            return cls(application=application, user_id=update.user_id)
 79        return super().from_update(update, application)
 80
 81
 82async def start(update: Update, context: CustomContext) -> None:
 83    """Display a message with instructions on how to use this bot."""
 84    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
 85    text = (
 86        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
 87        f"To post a custom update, call <code>{payload_url}</code>."
 88    )
 89    await update.message.reply_html(text=text)
 90
 91
 92async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
 93    """Handle custom updates."""
 94    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
 95    payloads = context.user_data.setdefault("payloads", [])
 96    payloads.append(update.payload)
 97    combined_payloads = "</code>\n• <code>".join(payloads)
 98    text = (
 99        f"The user {chat_member.user.mention_html()} has sent a new payload. "
100        f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
101    )
102    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
103
104
105async def telegram(request: HttpRequest) -> HttpResponse:
106    """Handle incoming Telegram updates by putting them into the `update_queue`"""
107    await ptb_application.update_queue.put(
108        Update.de_json(data=json.loads(request.body), bot=ptb_application.bot)
109    )
110    return HttpResponse()
111
112
113async def custom_updates(request: HttpRequest) -> HttpResponse:
114    """
115    Handle incoming webhook updates by also putting them into the `update_queue` if
116    the required parameters were passed correctly.
117    """
118    try:
119        user_id = int(request.GET["user_id"])
120        payload = request.GET["payload"]
121    except KeyError:
122        return HttpResponseBadRequest(
123            "Please pass both `user_id` and `payload` as query parameters.",
124        )
125    except ValueError:
126        return HttpResponseBadRequest("The `user_id` must be a string!")
127
128    await ptb_application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
129    return HttpResponse()
130
131
132async def health(_: HttpRequest) -> HttpResponse:
133    """For the health endpoint, reply with a simple plain text message."""
134    return HttpResponse("The bot is still running fine :)")
135
136
137# Set up PTB application and a web application for handling the incoming requests.
138
139context_types = ContextTypes(context=CustomContext)
140# Here we set updater to None because we want our custom webhook server to handle the updates
141# and hence we don't need an Updater instance
142ptb_application = (
143    Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
144)
145
146# register handlers
147ptb_application.add_handler(CommandHandler("start", start))
148ptb_application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
149
150urlpatterns = [
151    path("telegram", telegram, name="Telegram updates"),
152    path("submitpayload", custom_updates, name="custom updates"),
153    path("healthcheck", health, name="health check"),
154]
155settings.configure(ROOT_URLCONF=__name__, SECRET_KEY=uuid4().hex)
156
157
158async def main() -> None:
159    """Finalize configuration and run the applications."""
160    webserver = uvicorn.Server(
161        config=uvicorn.Config(
162            app=get_asgi_application(),
163            port=PORT,
164            use_colors=False,
165            host="127.0.0.1",
166        )
167    )
168
169    # Pass webhook settings to telegram
170    await ptb_application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
171
172    # Run application and webserver together
173    async with ptb_application:
174        await ptb_application.start()
175        await webserver.serve()
176        await ptb_application.stop()
177
178
179if __name__ == "__main__":
180    asyncio.run(main())