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"""
 16import asyncio
 17import html
 18import logging
 19from dataclasses import dataclass
 20from http import HTTPStatus
 21
 22import uvicorn
 23from starlette.applications import Starlette
 24from starlette.requests import Request
 25from starlette.responses import PlainTextResponse, Response
 26from starlette.routing import Route
 27
 28from telegram import Update
 29from telegram.constants import ParseMode
 30from telegram.ext import (
 31    Application,
 32    CallbackContext,
 33    CommandHandler,
 34    ContextTypes,
 35    ExtBot,
 36    TypeHandler,
 37)
 38
 39# Enable logging
 40logging.basicConfig(
 41    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
 42)
 43# set higher logging level for httpx to avoid all GET and POST requests being logged
 44logging.getLogger("httpx").setLevel(logging.WARNING)
 45
 46logger = logging.getLogger(__name__)
 47
 48# Define configuration constants
 49URL = "https://domain.tld"
 50ADMIN_CHAT_ID = 123456
 51PORT = 8000
 52TOKEN = "123:ABC"  # nosec B105
 53
 54
 55@dataclass
 56class WebhookUpdate:
 57    """Simple dataclass to wrap a custom update type"""
 58
 59    user_id: int
 60    payload: str
 61
 62
 63class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
 64    """
 65    Custom CallbackContext class that makes `user_data` available for updates of type
 66    `WebhookUpdate`.
 67    """
 68
 69    @classmethod
 70    def from_update(
 71        cls,
 72        update: object,
 73        application: "Application",
 74    ) -> "CustomContext":
 75        if isinstance(update, WebhookUpdate):
 76            return cls(application=application, user_id=update.user_id)
 77        return super().from_update(update, application)
 78
 79
 80async def start(update: Update, context: CustomContext) -> None:
 81    """Display a message with instructions on how to use this bot."""
 82    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
 83    text = (
 84        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
 85        f"To post a custom update, call <code>{payload_url}</code>."
 86    )
 87    await update.message.reply_html(text=text)
 88
 89
 90async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
 91    """Handle custom updates."""
 92    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
 93    payloads = context.user_data.setdefault("payloads", [])
 94    payloads.append(update.payload)
 95    combined_payloads = "</code>\n• <code>".join(payloads)
 96    text = (
 97        f"The user {chat_member.user.mention_html()} has sent a new payload. "
 98        f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
 99    )
100    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
101
102
103async def main() -> None:
104    """Set up PTB application and a web application for handling the incoming requests."""
105    context_types = ContextTypes(context=CustomContext)
106    # Here we set updater to None because we want our custom webhook server to handle the updates
107    # and hence we don't need an Updater instance
108    application = (
109        Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
110    )
111
112    # register handlers
113    application.add_handler(CommandHandler("start", start))
114    application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
115
116    # Pass webhook settings to telegram
117    await application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
118
119    # Set up webserver
120    async def telegram(request: Request) -> Response:
121        """Handle incoming Telegram updates by putting them into the `update_queue`"""
122        await application.update_queue.put(
123            Update.de_json(data=await request.json(), bot=application.bot)
124        )
125        return Response()
126
127    async def custom_updates(request: Request) -> PlainTextResponse:
128        """
129        Handle incoming webhook updates by also putting them into the `update_queue` if
130        the required parameters were passed correctly.
131        """
132        try:
133            user_id = int(request.query_params["user_id"])
134            payload = request.query_params["payload"]
135        except KeyError:
136            return PlainTextResponse(
137                status_code=HTTPStatus.BAD_REQUEST,
138                content="Please pass both `user_id` and `payload` as query parameters.",
139            )
140        except ValueError:
141            return PlainTextResponse(
142                status_code=HTTPStatus.BAD_REQUEST,
143                content="The `user_id` must be a string!",
144            )
145
146        await application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
147        return PlainTextResponse("Thank you for the submission! It's being forwarded.")
148
149    async def health(_: Request) -> PlainTextResponse:
150        """For the health endpoint, reply with a simple plain text message."""
151        return PlainTextResponse(content="The bot is still running fine :)")
152
153    starlette_app = Starlette(
154        routes=[
155            Route("/telegram", telegram, methods=["POST"]),
156            Route("/healthcheck", health, methods=["GET"]),
157            Route("/submitpayload", custom_updates, methods=["POST", "GET"]),
158        ]
159    )
160    webserver = uvicorn.Server(
161        config=uvicorn.Config(
162            app=starlette_app,
163            port=PORT,
164            use_colors=False,
165            host="127.0.0.1",
166        )
167    )
168
169    # Run application and webserver together
170    async with application:
171        await application.start()
172        await webserver.serve()
173        await application.stop()
174
175
176if __name__ == "__main__":
177    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"""
 16import asyncio
 17import html
 18import logging
 19from dataclasses import dataclass
 20from http import HTTPStatus
 21
 22import uvicorn
 23from asgiref.wsgi import WsgiToAsgi
 24from flask import Flask, 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    flask_app = Flask(__name__)
119
120    @flask_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(Update.de_json(data=request.json, bot=application.bot))
124        return Response(status=HTTPStatus.OK)
125
126    @flask_app.route("/submitpayload", methods=["GET", "POST"])  # type: ignore[misc]
127    async def custom_updates() -> Response:
128        """
129        Handle incoming webhook updates by also putting them into the `update_queue` if
130        the required parameters were passed correctly.
131        """
132        try:
133            user_id = int(request.args["user_id"])
134            payload = request.args["payload"]
135        except KeyError:
136            abort(
137                HTTPStatus.BAD_REQUEST,
138                "Please pass both `user_id` and `payload` as query parameters.",
139            )
140        except ValueError:
141            abort(HTTPStatus.BAD_REQUEST, "The `user_id` must be a string!")
142
143        await application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
144        return Response(status=HTTPStatus.OK)
145
146    @flask_app.get("/healthcheck")  # type: ignore[misc]
147    async def health() -> Response:
148        """For the health endpoint, reply with a simple plain text message."""
149        response = make_response("The bot is still running fine :)", HTTPStatus.OK)
150        response.mimetype = "text/plain"
151        return response
152
153    webserver = uvicorn.Server(
154        config=uvicorn.Config(
155            app=WsgiToAsgi(flask_app),
156            port=PORT,
157            use_colors=False,
158            host="127.0.0.1",
159        )
160    )
161
162    # Run application and webserver together
163    async with application:
164        await application.start()
165        await webserver.serve()
166        await application.stop()
167
168
169if __name__ == "__main__":
170    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"""
 16import asyncio
 17import html
 18import logging
 19from dataclasses import dataclass
 20from http import HTTPStatus
 21
 22import uvicorn
 23from quart import Quart, Response, abort, make_response, request
 24
 25from telegram import Update
 26from telegram.constants import ParseMode
 27from telegram.ext import (
 28    Application,
 29    CallbackContext,
 30    CommandHandler,
 31    ContextTypes,
 32    ExtBot,
 33    TypeHandler,
 34)
 35
 36# Enable logging
 37logging.basicConfig(
 38    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
 39)
 40# set higher logging level for httpx to avoid all GET and POST requests being logged
 41logging.getLogger("httpx").setLevel(logging.WARNING)
 42
 43logger = logging.getLogger(__name__)
 44
 45# Define configuration constants
 46URL = "https://domain.tld"
 47ADMIN_CHAT_ID = 123456
 48PORT = 8000
 49TOKEN = "123:ABC"  # nosec B105
 50
 51
 52@dataclass
 53class WebhookUpdate:
 54    """Simple dataclass to wrap a custom update type"""
 55
 56    user_id: int
 57    payload: str
 58
 59
 60class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
 61    """
 62    Custom CallbackContext class that makes `user_data` available for updates of type
 63    `WebhookUpdate`.
 64    """
 65
 66    @classmethod
 67    def from_update(
 68        cls,
 69        update: object,
 70        application: "Application",
 71    ) -> "CustomContext":
 72        if isinstance(update, WebhookUpdate):
 73            return cls(application=application, user_id=update.user_id)
 74        return super().from_update(update, application)
 75
 76
 77async def start(update: Update, context: CustomContext) -> None:
 78    """Display a message with instructions on how to use this bot."""
 79    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
 80    text = (
 81        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
 82        f"To post a custom update, call <code>{payload_url}</code>."
 83    )
 84    await update.message.reply_html(text=text)
 85
 86
 87async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
 88    """Handle custom updates."""
 89    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
 90    payloads = context.user_data.setdefault("payloads", [])
 91    payloads.append(update.payload)
 92    combined_payloads = "</code>\n• <code>".join(payloads)
 93    text = (
 94        f"The user {chat_member.user.mention_html()} has sent a new payload. "
 95        f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
 96    )
 97    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
 98
 99
100async def main() -> None:
101    """Set up PTB application and a web application for handling the incoming requests."""
102    context_types = ContextTypes(context=CustomContext)
103    # Here we set updater to None because we want our custom webhook server to handle the updates
104    # and hence we don't need an Updater instance
105    application = (
106        Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
107    )
108
109    # register handlers
110    application.add_handler(CommandHandler("start", start))
111    application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
112
113    # Pass webhook settings to telegram
114    await application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
115
116    # Set up webserver
117    quart_app = Quart(__name__)
118
119    @quart_app.post("/telegram")  # type: ignore[misc]
120    async def telegram() -> Response:
121        """Handle incoming Telegram updates by putting them into the `update_queue`"""
122        await application.update_queue.put(
123            Update.de_json(data=await request.get_json(), bot=application.bot)
124        )
125        return Response(status=HTTPStatus.OK)
126
127    @quart_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    @quart_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 = await 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=quart_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 `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"""
 16import asyncio
 17import html
 18import json
 19import logging
 20from dataclasses import dataclass
 21from uuid import uuid4
 22
 23import uvicorn
 24from django.conf import settings
 25from django.core.asgi import get_asgi_application
 26from django.http import HttpRequest, HttpResponse, HttpResponseBadRequest
 27from django.urls import path
 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 telegram(request: HttpRequest) -> HttpResponse:
105    """Handle incoming Telegram updates by putting them into the `update_queue`"""
106    await ptb_application.update_queue.put(
107        Update.de_json(data=json.loads(request.body), bot=ptb_application.bot)
108    )
109    return HttpResponse()
110
111
112async def custom_updates(request: HttpRequest) -> HttpResponse:
113    """
114    Handle incoming webhook updates by also putting them into the `update_queue` if
115    the required parameters were passed correctly.
116    """
117    try:
118        user_id = int(request.GET["user_id"])
119        payload = request.GET["payload"]
120    except KeyError:
121        return HttpResponseBadRequest(
122            "Please pass both `user_id` and `payload` as query parameters.",
123        )
124    except ValueError:
125        return HttpResponseBadRequest("The `user_id` must be a string!")
126
127    await ptb_application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
128    return HttpResponse()
129
130
131async def health(_: HttpRequest) -> HttpResponse:
132    """For the health endpoint, reply with a simple plain text message."""
133    return HttpResponse("The bot is still running fine :)")
134
135
136# Set up PTB application and a web application for handling the incoming requests.
137
138context_types = ContextTypes(context=CustomContext)
139# Here we set updater to None because we want our custom webhook server to handle the updates
140# and hence we don't need an Updater instance
141ptb_application = (
142    Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
143)
144
145# register handlers
146ptb_application.add_handler(CommandHandler("start", start))
147ptb_application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
148
149urlpatterns = [
150    path("telegram", telegram, name="Telegram updates"),
151    path("submitpayload", custom_updates, name="custom updates"),
152    path("healthcheck", health, name="health check"),
153]
154settings.configure(ROOT_URLCONF=__name__, SECRET_KEY=uuid4().hex)
155
156
157async def main() -> None:
158    """Finalize configuration and run the applications."""
159    webserver = uvicorn.Server(
160        config=uvicorn.Config(
161            app=get_asgi_application(),
162            port=PORT,
163            use_colors=False,
164            host="127.0.0.1",
165        )
166    )
167
168    # Pass webhook settings to telegram
169    await ptb_application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
170
171    # Run application and webserver together
172    async with ptb_application:
173        await ptb_application.start()
174        await webserver.serve()
175        await ptb_application.stop()
176
177
178if __name__ == "__main__":
179    asyncio.run(main())