Hide navigation sidebar
Hide table of contents sidebar
Skip to content
Back to top
View this page
Toggle table of contents sidebar

# `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!

`starlette`

      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())

`flask`

      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())

`quart`

      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())

`Django`

      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())