Skip to content
Back to top
View this page

# `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"""
     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())

`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"""
     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[untyped-decorator]
    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[untyped-decorator]
    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[untyped-decorator]
    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())

`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"""
     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[untyped-decorator]
    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[untyped-decorator]
    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[untyped-decorator]
    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())

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