Provide Versions of customwebhookbot.py with Different Frameworks (#3820)

docs/requirements-docs.txt

@@ -5,3 +5,4 @@ git+https://github.com/harshil21/furo-sphinx-search@01efc7be422d7dc02390aab9be68
 sphinx-paramlinks==0.5.4
 sphinxcontrib-mermaid==0.9.2
 sphinx-copybutton==0.5.2
+sphinx-inline-tabs==2023.4.21

docs/source/conf.py

@@ -39,6 +39,7 @@ extensions = [
     "sphinx.ext.extlinks",
     "sphinx_paramlinks",
     "sphinx_copybutton",
+    "sphinx_inline_tabs",
     "sphinxcontrib.mermaid",
     "sphinx_search.extension",
 ]

docs/source/examples.customwebhookbot.rst

@@ -1,7 +1,43 @@
 ``customwebhookbot.py``
 =======================
 
-.. literalinclude:: ../../examples/customwebhookbot.py
-   :language: python
-   :linenos:
+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!
+
+.. tab:: ``starlette``
+
+    .. literalinclude:: ../../examples/customwebhookbot/starlettebot.py
+       :language: python
+       :linenos:
+
+.. tab:: ``flask``
+
+    .. literalinclude:: ../../examples/customwebhookbot/flaskbot.py
+       :language: python
+       :linenos:
+
+.. tab:: ``quart``
+
+    .. literalinclude:: ../../examples/customwebhookbot/quartbot.py
+       :language: python
+       :linenos:
+
+.. tab:: ``Django``
+
+    .. literalinclude:: ../../examples/customwebhookbot/djangobot.py
+       :language: python
+       :linenos:
     

examples/customwebhookbot/djangobot.py

@@ -0,0 +1,193 @@
+#!/usr/bin/env python
+# This program is dedicated to the public domain under the CC0 license.
+# pylint: disable=import-error,wrong-import-position,unused-argument
+"""
+Simple example of a bot that uses a custom webhook setup and handles custom updates.
+For the custom webhook setup, the libraries `Django` and `uvicorn` are used. Please
+install them as `pip install Django~=4.2.4 uvicorn~=0.23.2`.
+Note that any other `asyncio` based web server framework can be used for a custom webhook setup
+just as well.
+
+Usage:
+Set bot Token, URL, admin CHAT_ID and PORT after the imports.
+You may also need to change the `listen` value in the uvicorn configuration to match your setup.
+Press Ctrl-C on the command line or send a signal to the process to stop the bot.
+"""
+import asyncio
+import html
+import json
+import logging
+from dataclasses import dataclass
+from uuid import uuid4
+
+import uvicorn
+from django.conf import settings
+from django.core.asgi import get_asgi_application
+from django.http import HttpRequest, HttpResponse, HttpResponseBadRequest
+from django.urls import path
+
+from telegram import __version__ as TG_VER
+
+try:
+    from telegram import __version_info__
+except ImportError:
+    __version_info__ = (0, 0, 0, 0, 0)  # type: ignore[assignment]
+
+if __version_info__ < (20, 0, 0, "alpha", 1):
+    raise RuntimeError(
+        f"This example is not compatible with your current PTB version {TG_VER}. To view the "
+        f"{TG_VER} version of this example, "
+        f"visit https://docs.python-telegram-bot.org/en/v{TG_VER}/examples.html"
+    )
+
+from telegram import Update
+from telegram.constants import ParseMode
+from telegram.ext import (
+    Application,
+    CallbackContext,
+    CommandHandler,
+    ContextTypes,
+    ExtBot,
+    TypeHandler,
+)
+
+# Enable logging
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
+)
+# set higher logging level for httpx to avoid all GET and POST requests being logged
+logging.getLogger("httpx").setLevel(logging.WARNING)
+
+logger = logging.getLogger(__name__)
+
+# Define configuration constants
+URL = "https://domain.tld"
+ADMIN_CHAT_ID = 123456
+PORT = 8000
+TOKEN = "123:ABC"  # nosec B105
+
+
+@dataclass
+class WebhookUpdate:
+    """Simple dataclass to wrap a custom update type"""
+
+    user_id: int
+    payload: str
+
+
+class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
+    """
+    Custom CallbackContext class that makes `user_data` available for updates of type
+    `WebhookUpdate`.
+    """
+
+    @classmethod
+    def from_update(
+        cls,
+        update: object,
+        application: "Application",
+    ) -> "CustomContext":
+        if isinstance(update, WebhookUpdate):
+            return cls(application=application, user_id=update.user_id)
+        return super().from_update(update, application)
+
+
+async def start(update: Update, context: CustomContext) -> None:
+    """Display a message with instructions on how to use this bot."""
+    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
+    text = (
+        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
+        f"To post a custom update, call <code>{payload_url}</code>."
+    )
+    await update.message.reply_html(text=text)
+
+
+async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
+    """Handle custom updates."""
+    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
+    payloads = context.user_data.setdefault("payloads", [])
+    payloads.append(update.payload)
+    combined_payloads = "</code>\n• <code>".join(payloads)
+    text = (
+        f"The user {chat_member.user.mention_html()} has sent a new payload. "
+        f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
+    )
+    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
+
+
+async def telegram(request: HttpRequest) -> HttpResponse:
+    """Handle incoming Telegram updates by putting them into the `update_queue`"""
+    await ptb_application.update_queue.put(
+        Update.de_json(data=json.loads(request.body), bot=ptb_application.bot)
+    )
+    return HttpResponse()
+
+
+async def custom_updates(request: HttpRequest) -> HttpResponse:
+    """
+    Handle incoming webhook updates by also putting them into the `update_queue` if
+    the required parameters were passed correctly.
+    """
+    try:
+        user_id = int(request.GET["user_id"])
+        payload = request.GET["payload"]
+    except KeyError:
+        return HttpResponseBadRequest(
+            "Please pass both `user_id` and `payload` as query parameters.",
+        )
+    except ValueError:
+        return HttpResponseBadRequest("The `user_id` must be a string!")
+
+    await ptb_application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
+    return HttpResponse()
+
+
+async def health(_: HttpRequest) -> HttpResponse:
+    """For the health endpoint, reply with a simple plain text message."""
+    return HttpResponse("The bot is still running fine :)")
+
+
+# Set up PTB application and a web application for handling the incoming requests.
+
+context_types = ContextTypes(context=CustomContext)
+# Here we set updater to None because we want our custom webhook server to handle the updates
+# and hence we don't need an Updater instance
+ptb_application = (
+    Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
+)
+
+# register handlers
+ptb_application.add_handler(CommandHandler("start", start))
+ptb_application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
+
+urlpatterns = [
+    path("telegram", telegram, name="Telegram updates"),
+    path("submitpayload", custom_updates, name="custom updates"),
+    path("healthcheck", health, name="health check"),
+]
+settings.configure(ROOT_URLCONF=__name__, SECRET_KEY=uuid4().hex)
+
+
+async def main() -> None:
+    """Finalize configuration and run the applications."""
+    webserver = uvicorn.Server(
+        config=uvicorn.Config(
+            app=get_asgi_application(),
+            port=PORT,
+            use_colors=False,
+            host="127.0.0.1",
+        )
+    )
+
+    # Pass webhook settings to telegram
+    await ptb_application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
+
+    # Run application and webserver together
+    async with ptb_application:
+        await ptb_application.start()
+        await webserver.serve()
+        await ptb_application.stop()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/customwebhookbot/flaskbot.py

@@ -0,0 +1,184 @@
+#!/usr/bin/env python
+# This program is dedicated to the public domain under the CC0 license.
+# pylint: disable=import-error,wrong-import-position,unused-argument
+"""
+Simple example of a bot that uses a custom webhook setup and handles custom updates.
+For the custom webhook setup, the libraries `flask`, `asgiref` and `uvicorn` are used. Please
+install them as `pip install flask[async]~=2.3.2 uvicorn~=0.23.2 asgiref~=3.7.2`.
+Note that any other `asyncio` based web server framework can be used for a custom webhook setup
+just as well.
+
+Usage:
+Set bot Token, URL, admin CHAT_ID and PORT after the imports.
+You may also need to change the `listen` value in the uvicorn configuration to match your setup.
+Press Ctrl-C on the command line or send a signal to the process to stop the bot.
+"""
+import asyncio
+import html
+import logging
+from dataclasses import dataclass
+from http import HTTPStatus
+
+import uvicorn
+from asgiref.wsgi import WsgiToAsgi
+from flask import Flask, Response, abort, make_response, request
+
+from telegram import __version__ as TG_VER
+
+try:
+    from telegram import __version_info__
+except ImportError:
+    __version_info__ = (0, 0, 0, 0, 0)  # type: ignore[assignment]
+
+if __version_info__ < (20, 0, 0, "alpha", 1):
+    raise RuntimeError(
+        f"This example is not compatible with your current PTB version {TG_VER}. To view the "
+        f"{TG_VER} version of this example, "
+        f"visit https://docs.python-telegram-bot.org/en/v{TG_VER}/examples.html"
+    )
+
+from telegram import Update
+from telegram.constants import ParseMode
+from telegram.ext import (
+    Application,
+    CallbackContext,
+    CommandHandler,
+    ContextTypes,
+    ExtBot,
+    TypeHandler,
+)
+
+# Enable logging
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
+)
+# set higher logging level for httpx to avoid all GET and POST requests being logged
+logging.getLogger("httpx").setLevel(logging.WARNING)
+
+logger = logging.getLogger(__name__)
+
+# Define configuration constants
+URL = "https://domain.tld"
+ADMIN_CHAT_ID = 123456
+PORT = 8000
+TOKEN = "123:ABC"  # nosec B105
+
+
+@dataclass
+class WebhookUpdate:
+    """Simple dataclass to wrap a custom update type"""
+
+    user_id: int
+    payload: str
+
+
+class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
+    """
+    Custom CallbackContext class that makes `user_data` available for updates of type
+    `WebhookUpdate`.
+    """
+
+    @classmethod
+    def from_update(
+        cls,
+        update: object,
+        application: "Application",
+    ) -> "CustomContext":
+        if isinstance(update, WebhookUpdate):
+            return cls(application=application, user_id=update.user_id)
+        return super().from_update(update, application)
+
+
+async def start(update: Update, context: CustomContext) -> None:
+    """Display a message with instructions on how to use this bot."""
+    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
+    text = (
+        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
+        f"To post a custom update, call <code>{payload_url}</code>."
+    )
+    await update.message.reply_html(text=text)
+
+
+async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
+    """Handle custom updates."""
+    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
+    payloads = context.user_data.setdefault("payloads", [])
+    payloads.append(update.payload)
+    combined_payloads = "</code>\n• <code>".join(payloads)
+    text = (
+        f"The user {chat_member.user.mention_html()} has sent a new payload. "
+        f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
+    )
+    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
+
+
+async def main() -> None:
+    """Set up PTB application and a web application for handling the incoming requests."""
+    context_types = ContextTypes(context=CustomContext)
+    # Here we set updater to None because we want our custom webhook server to handle the updates
+    # and hence we don't need an Updater instance
+    application = (
+        Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
+    )
+
+    # register handlers
+    application.add_handler(CommandHandler("start", start))
+    application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
+
+    # Pass webhook settings to telegram
+    await application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
+
+    # Set up webserver
+    flask_app = Flask(__name__)
+
+    @flask_app.post("/telegram")  # type: ignore[misc]
+    async def telegram() -> Response:
+        """Handle incoming Telegram updates by putting them into the `update_queue`"""
+        await application.update_queue.put(Update.de_json(data=request.json, bot=application.bot))
+        return Response(status=HTTPStatus.OK)
+
+    @flask_app.route("/submitpayload", methods=["GET", "POST"])  # type: ignore[misc]
+    async def custom_updates() -> Response:
+        """
+        Handle incoming webhook updates by also putting them into the `update_queue` if
+        the required parameters were passed correctly.
+        """
+        try:
+            user_id = int(request.args["user_id"])
+            payload = request.args["payload"]
+        except KeyError:
+            abort(
+                HTTPStatus.BAD_REQUEST,
+                "Please pass both `user_id` and `payload` as query parameters.",
+            )
+        except ValueError:
+            abort(HTTPStatus.BAD_REQUEST, "The `user_id` must be a string!")
+
+        await application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
+        return Response(status=HTTPStatus.OK)
+
+    @flask_app.get("/healthcheck")  # type: ignore[misc]
+    async def health() -> Response:
+        """For the health endpoint, reply with a simple plain text message."""
+        response = make_response("The bot is still running fine :)", HTTPStatus.OK)
+        response.mimetype = "text/plain"
+        return response
+
+    webserver = uvicorn.Server(
+        config=uvicorn.Config(
+            app=WsgiToAsgi(flask_app),
+            port=PORT,
+            use_colors=False,
+            host="127.0.0.1",
+        )
+    )
+
+    # Run application and webserver together
+    async with application:
+        await application.start()
+        await webserver.serve()
+        await application.stop()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/customwebhookbot/quartbot.py

@@ -0,0 +1,185 @@
+#!/usr/bin/env python
+# This program is dedicated to the public domain under the CC0 license.
+# pylint: disable=import-error,wrong-import-position,unused-argument
+"""
+Simple example of a bot that uses a custom webhook setup and handles custom updates.
+For the custom webhook setup, the libraries `quart` and `uvicorn` are used. Please
+install them as `pip install quart~=0.18.4 uvicorn~=0.23.2`.
+Note that any other `asyncio` based web server framework can be used for a custom webhook setup
+just as well.
+
+Usage:
+Set bot Token, URL, admin CHAT_ID and PORT after the imports.
+You may also need to change the `listen` value in the uvicorn configuration to match your setup.
+Press Ctrl-C on the command line or send a signal to the process to stop the bot.
+"""
+import asyncio
+import html
+import logging
+from dataclasses import dataclass
+from http import HTTPStatus
+
+import uvicorn
+from quart import Quart, Response, abort, make_response, request
+
+from telegram import __version__ as TG_VER
+
+try:
+    from telegram import __version_info__
+except ImportError:
+    __version_info__ = (0, 0, 0, 0, 0)  # type: ignore[assignment]
+
+if __version_info__ < (20, 0, 0, "alpha", 1):
+    raise RuntimeError(
+        f"This example is not compatible with your current PTB version {TG_VER}. To view the "
+        f"{TG_VER} version of this example, "
+        f"visit https://docs.python-telegram-bot.org/en/v{TG_VER}/examples.html"
+    )
+
+from telegram import Update
+from telegram.constants import ParseMode
+from telegram.ext import (
+    Application,
+    CallbackContext,
+    CommandHandler,
+    ContextTypes,
+    ExtBot,
+    TypeHandler,
+)
+
+# Enable logging
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
+)
+# set higher logging level for httpx to avoid all GET and POST requests being logged
+logging.getLogger("httpx").setLevel(logging.WARNING)
+
+logger = logging.getLogger(__name__)
+
+# Define configuration constants
+URL = "https://domain.tld"
+ADMIN_CHAT_ID = 123456
+PORT = 8000
+TOKEN = "123:ABC"  # nosec B105
+
+
+@dataclass
+class WebhookUpdate:
+    """Simple dataclass to wrap a custom update type"""
+
+    user_id: int
+    payload: str
+
+
+class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
+    """
+    Custom CallbackContext class that makes `user_data` available for updates of type
+    `WebhookUpdate`.
+    """
+
+    @classmethod
+    def from_update(
+        cls,
+        update: object,
+        application: "Application",
+    ) -> "CustomContext":
+        if isinstance(update, WebhookUpdate):
+            return cls(application=application, user_id=update.user_id)
+        return super().from_update(update, application)
+
+
+async def start(update: Update, context: CustomContext) -> None:
+    """Display a message with instructions on how to use this bot."""
+    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
+    text = (
+        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
+        f"To post a custom update, call <code>{payload_url}</code>."
+    )
+    await update.message.reply_html(text=text)
+
+
+async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
+    """Handle custom updates."""
+    chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
+    payloads = context.user_data.setdefault("payloads", [])
+    payloads.append(update.payload)
+    combined_payloads = "</code>\n• <code>".join(payloads)
+    text = (
+        f"The user {chat_member.user.mention_html()} has sent a new payload. "
+        f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
+    )
+    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
+
+
+async def main() -> None:
+    """Set up PTB application and a web application for handling the incoming requests."""
+    context_types = ContextTypes(context=CustomContext)
+    # Here we set updater to None because we want our custom webhook server to handle the updates
+    # and hence we don't need an Updater instance
+    application = (
+        Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
+    )
+
+    # register handlers
+    application.add_handler(CommandHandler("start", start))
+    application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
+
+    # Pass webhook settings to telegram
+    await application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
+
+    # Set up webserver
+    quart_app = Quart(__name__)
+
+    @quart_app.post("/telegram")  # type: ignore[misc]
+    async def telegram() -> Response:
+        """Handle incoming Telegram updates by putting them into the `update_queue`"""
+        await application.update_queue.put(
+            Update.de_json(data=await request.get_json(), bot=application.bot)
+        )
+        return Response(status=HTTPStatus.OK)
+
+    @quart_app.route("/submitpayload", methods=["GET", "POST"])  # type: ignore[misc]
+    async def custom_updates() -> Response:
+        """
+        Handle incoming webhook updates by also putting them into the `update_queue` if
+        the required parameters were passed correctly.
+        """
+        try:
+            user_id = int(request.args["user_id"])
+            payload = request.args["payload"]
+        except KeyError:
+            abort(
+                HTTPStatus.BAD_REQUEST,
+                "Please pass both `user_id` and `payload` as query parameters.",
+            )
+        except ValueError:
+            abort(HTTPStatus.BAD_REQUEST, "The `user_id` must be a string!")
+
+        await application.update_queue.put(WebhookUpdate(user_id=user_id, payload=payload))
+        return Response(status=HTTPStatus.OK)
+
+    @quart_app.get("/healthcheck")  # type: ignore[misc]
+    async def health() -> Response:
+        """For the health endpoint, reply with a simple plain text message."""
+        response = await make_response("The bot is still running fine :)", HTTPStatus.OK)
+        response.mimetype = "text/plain"
+        return response
+
+    webserver = uvicorn.Server(
+        config=uvicorn.Config(
+            app=quart_app,
+            port=PORT,
+            use_colors=False,
+            host="127.0.0.1",
+        )
+    )
+
+    # Run application and webserver together
+    async with application:
+        await application.start()
+        await webserver.serve()
+        await application.stop()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/customwebhookbot/starlettebot.py

@@ -1,15 +1,15 @@
 #!/usr/bin/env python
 # This program is dedicated to the public domain under the CC0 license.
-# pylint: disable=import-error,wrong-import-position
+# pylint: disable=import-error,wrong-import-position,unused-argument
 """
 Simple example of a bot that uses a custom webhook setup and handles custom updates.
 For the custom webhook setup, the libraries `starlette` and `uvicorn` are used. Please install
-them as `pip install starlette~=0.20.0 uvicorn~=0.17.0`.
+them as `pip install starlette~=0.20.0 uvicorn~=0.23.2`.
 Note that any other `asyncio` based web server framework can be used for a custom webhook setup
 just as well.
 
 Usage:
-Set bot token, url, admin chat_id and port at the start of the `main` function.
+Set bot Token, URL, admin CHAT_ID and PORT after the imports.
 You may also need to change the `listen` value in the uvicorn configuration to match your setup.
 Press Ctrl-C on the command line or send a signal to the process to stop the bot.
 """
@@ -59,6 +59,12 @@ logging.getLogger("httpx").setLevel(logging.WARNING)
 
 logger = logging.getLogger(__name__)
 
+# Define configuration constants
+URL = "https://domain.tld"
+ADMIN_CHAT_ID = 123456
+PORT = 8000
+TOKEN = "123:ABC"  # nosec B105
+
 
 @dataclass
 class WebhookUpdate:
@@ -87,17 +93,16 @@ class CustomContext(CallbackContext[ExtBot, dict, dict, dict]):
 
 async def start(update: Update, context: CustomContext) -> None:
     """Display a message with instructions on how to use this bot."""
-    url = context.bot_data["url"]
-    payload_url = html.escape(f"{url}/submitpayload?user_id=<your user id>&payload=<payload>")
+    payload_url = html.escape(f"{URL}/submitpayload?user_id=<your user id>&payload=<payload>")
     text = (
-        f"To check if the bot is still running, call <code>{url}/healthcheck</code>.\n\n"
+        f"To check if the bot is still running, call <code>{URL}/healthcheck</code>.\n\n"
         f"To post a custom update, call <code>{payload_url}</code>."
     )
     await update.message.reply_html(text=text)
 
 
 async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
-    """Callback that handles the custom updates."""
+    """Handle custom updates."""
     chat_member = await context.bot.get_chat_member(chat_id=update.user_id, user_id=update.user_id)
     payloads = context.user_data.setdefault("payloads", [])
     payloads.append(update.payload)
@@ -106,33 +111,24 @@ async def webhook_update(update: WebhookUpdate, context: CustomContext) -> None:
         f"The user {chat_member.user.mention_html()} has sent a new payload. "
         f"So far they have sent the following payloads: \n\n• <code>{combined_payloads}</code>"
     )
-    await context.bot.send_message(
-        chat_id=context.bot_data["admin_chat_id"], text=text, parse_mode=ParseMode.HTML
-    )
+    await context.bot.send_message(chat_id=ADMIN_CHAT_ID, text=text, parse_mode=ParseMode.HTML)
 
 
 async def main() -> None:
-    """Set up the application and a custom webserver."""
-    url = "https://domain.tld"
-    admin_chat_id = 123456
-    port = 8000
-
+    """Set up PTB application and a web application for handling the incoming requests."""
     context_types = ContextTypes(context=CustomContext)
     # Here we set updater to None because we want our custom webhook server to handle the updates
     # and hence we don't need an Updater instance
     application = (
-        Application.builder().token("TOKEN").updater(None).context_types(context_types).build()
+        Application.builder().token(TOKEN).updater(None).context_types(context_types).build()
     )
-    # save the values in `bot_data` such that we may easily access them in the callbacks
-    application.bot_data["url"] = url
-    application.bot_data["admin_chat_id"] = admin_chat_id
 
     # register handlers
     application.add_handler(CommandHandler("start", start))
     application.add_handler(TypeHandler(type=WebhookUpdate, callback=webhook_update))
 
     # Pass webhook settings to telegram
-    await application.bot.set_webhook(url=f"{url}/telegram", allowed_updates=Update.ALL_TYPES)
+    await application.bot.set_webhook(url=f"{URL}/telegram", allowed_updates=Update.ALL_TYPES)
 
     # Set up webserver
     async def telegram(request: Request) -> Response:
@@ -178,7 +174,7 @@ async def main() -> None:
     webserver = uvicorn.Server(
         config=uvicorn.Config(
             app=starlette_app,
-            port=port,
+            port=PORT,
             use_colors=False,
             host="127.0.0.1",
         )

setup.cfg

@@ -78,9 +78,17 @@ warn_unused_ignores = False
 [mypy-apscheduler.*]
 ignore_missing_imports = True
 
-# uvicorn and starlette are only used for the `customwebhookbot.py` example
+# The libs listed below are only used for the `customwebhookbot_*.py` examples
 # let's just ignore type checking for them for now
 [mypy-uvicorn.*]
 ignore_missing_imports = True
 [mypy-starlette.*]
 ignore_missing_imports = True
+[mypy-asgiref.*]
+ignore_missing_imports = True
+[mypy-flask.*]
+ignore_missing_imports = True
+[mypy-quart.*]
+ignore_missing_imports = True
+[mypy-django.*]
+ignore_missing_imports = True