We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support telegram-bot v20+
### Body

Currently our telegram integration uses Telegram v13 telegram-bot library. On 1st of Jan 2023 a new, backwards incompatible version of Telegram-bot has been released : https://pypi.org/project/python-telegram-bot/20.0/#history and at least as reported by MyPy and our test suite test failures, Telegram 20 needs some changes to work:

Here is a transition guide that might be helpful. 

Transition guide is here: https://github.com/python-telegram-bot/python-telegram-bot/wiki/Transition-guide-to-Version-20.0

In the meantime we limit telegram to < 20.0.0

### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/providers/telegram/hooks/telegram.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 """Hook for Telegram"""
19 from __future__ import annotations
20 
21 import telegram
22 import tenacity
23 
24 from airflow.exceptions import AirflowException
25 from airflow.hooks.base import BaseHook
26 
27 
28 class TelegramHook(BaseHook):
29     """
30     This hook allows you to post messages to Telegram using the telegram python-telegram-bot library.
31 
32     The library can be found here: https://github.com/python-telegram-bot/python-telegram-bot
33     It accepts both telegram bot API token directly or connection that has telegram bot API token.
34     If both supplied, token parameter will be given precedence, otherwise 'password' field in the connection
35     from telegram_conn_id will be used.
36     chat_id can also be provided in the connection using 'host' field in connection.
37     Following is the details of a telegram_connection:
38     name: 'telegram-connection-name'
39     conn_type: 'http'
40     password: 'TELEGRAM_TOKEN'
41     host: 'chat_id' (optional)
42     Examples:
43     .. code-block:: python
44 
45         # Create hook
46         telegram_hook = TelegramHook(telegram_conn_id="telegram_default")
47         # or telegram_hook = TelegramHook(telegram_conn_id='telegram_default', chat_id='-1xxx')
48         # or telegram_hook = TelegramHook(token='xxx:xxx', chat_id='-1xxx')
49 
50         # Call method from telegram bot client
51         telegram_hook.send_message(None, {"text": "message", "chat_id": "-1xxx"})
52         # or telegram_hook.send_message(None', {"text": "message"})
53 
54     :param telegram_conn_id: connection that optionally has Telegram API token in the password field
55     :param token: optional telegram API token
56     :param chat_id: optional chat_id of the telegram chat/channel/group
57     """
58 
59     def __init__(
60         self,
61         telegram_conn_id: str | None = None,
62         token: str | None = None,
63         chat_id: str | None = None,
64     ) -> None:
65         super().__init__()
66         self.token = self.__get_token(token, telegram_conn_id)
67         self.chat_id = self.__get_chat_id(chat_id, telegram_conn_id)
68         self.connection = self.get_conn()
69 
70     def get_conn(self) -> telegram.bot.Bot:
71         """
72         Returns the telegram bot client
73 
74         :return: telegram bot client
75         """
76         return telegram.bot.Bot(token=self.token)
77 
78     def __get_token(self, token: str | None, telegram_conn_id: str | None) -> str:
79         """
80         Returns the telegram API token
81 
82         :param token: telegram API token
83         :param telegram_conn_id: telegram connection name
84         :return: telegram API token
85         """
86         if token is not None:
87             return token
88 
89         if telegram_conn_id is not None:
90             conn = self.get_connection(telegram_conn_id)
91 
92             if not conn.password:
93                 raise AirflowException("Missing token(password) in Telegram connection")
94 
95             return conn.password
96 
97         raise AirflowException("Cannot get token: No valid Telegram connection supplied.")
98 
99     def __get_chat_id(self, chat_id: str | None, telegram_conn_id: str | None) -> str | None:
100         """
101         Returns the telegram chat ID for a chat/channel/group
102 
103         :param chat_id: optional chat ID
104         :param telegram_conn_id: telegram connection name
105         :return: telegram chat ID
106         """
107         if chat_id is not None:
108             return chat_id
109 
110         if telegram_conn_id is not None:
111             conn = self.get_connection(telegram_conn_id)
112             return conn.host
113 
114         return None
115 
116     @tenacity.retry(
117         retry=tenacity.retry_if_exception_type(telegram.error.TelegramError),
118         stop=tenacity.stop_after_attempt(5),
119         wait=tenacity.wait_fixed(1),
120     )
121     def send_message(self, api_params: dict) -> None:
122         """
123         Sends the message to a telegram channel or chat.
124 
125         :param api_params: params for telegram_instance.send_message. It can also be used to override chat_id
126         """
127         kwargs = {
128             "chat_id": self.chat_id,
129             "parse_mode": telegram.parsemode.ParseMode.HTML,
130             "disable_web_page_preview": True,
131         }
132         kwargs.update(api_params)
133 
134         if "text" not in kwargs or kwargs["text"] is None:
135             raise AirflowException("'text' must be provided for telegram message")
136 
137         if kwargs["chat_id"] is None:
138             raise AirflowException("'chat_id' must be provided for telegram message")
139 
140         response = self.connection.send_message(**kwargs)
141         self.log.debug(response)
142
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/providers/telegram/hooks/telegram.py b/airflow/providers/telegram/hooks/telegram.py
--- a/airflow/providers/telegram/hooks/telegram.py
+++ b/airflow/providers/telegram/hooks/telegram.py
@@ -18,6 +18,8 @@
 """Hook for Telegram"""
 from __future__ import annotations
 
+import asyncio
+
 import telegram
 import tenacity
 
@@ -67,13 +69,13 @@
         self.chat_id = self.__get_chat_id(chat_id, telegram_conn_id)
         self.connection = self.get_conn()
 
-    def get_conn(self) -> telegram.bot.Bot:
+    def get_conn(self) -> telegram.Bot:
         """
         Returns the telegram bot client
 
         :return: telegram bot client
         """
-        return telegram.bot.Bot(token=self.token)
+        return telegram.Bot(self.token)
 
     def __get_token(self, token: str | None, telegram_conn_id: str | None) -> str:
         """
@@ -126,7 +128,7 @@
         """
         kwargs = {
             "chat_id": self.chat_id,
-            "parse_mode": telegram.parsemode.ParseMode.HTML,
+            "parse_mode": telegram.constants.ParseMode.HTML,
             "disable_web_page_preview": True,
         }
         kwargs.update(api_params)
@@ -137,5 +139,5 @@
         if kwargs["chat_id"] is None:
             raise AirflowException("'chat_id' must be provided for telegram message")
 
-        response = self.connection.send_message(**kwargs)
+        response = asyncio.run(self.connection.send_message(**kwargs))
         self.log.debug(response)

{"golden_diff": "diff --git a/airflow/providers/telegram/hooks/telegram.py b/airflow/providers/telegram/hooks/telegram.py\n--- a/airflow/providers/telegram/hooks/telegram.py\n+++ b/airflow/providers/telegram/hooks/telegram.py\n@@ -18,6 +18,8 @@\n \"\"\"Hook for Telegram\"\"\"\n from __future__ import annotations\n \n+import asyncio\n+\n import telegram\n import tenacity\n \n@@ -67,13 +69,13 @@\n         self.chat_id = self.__get_chat_id(chat_id, telegram_conn_id)\n         self.connection = self.get_conn()\n \n-    def get_conn(self) -> telegram.bot.Bot:\n+    def get_conn(self) -> telegram.Bot:\n         \"\"\"\n         Returns the telegram bot client\n \n         :return: telegram bot client\n         \"\"\"\n-        return telegram.bot.Bot(token=self.token)\n+        return telegram.Bot(self.token)\n \n     def __get_token(self, token: str | None, telegram_conn_id: str | None) -> str:\n         \"\"\"\n@@ -126,7 +128,7 @@\n         \"\"\"\n         kwargs = {\n             \"chat_id\": self.chat_id,\n-            \"parse_mode\": telegram.parsemode.ParseMode.HTML,\n+            \"parse_mode\": telegram.constants.ParseMode.HTML,\n             \"disable_web_page_preview\": True,\n         }\n         kwargs.update(api_params)\n@@ -137,5 +139,5 @@\n         if kwargs[\"chat_id\"] is None:\n             raise AirflowException(\"'chat_id' must be provided for telegram message\")\n \n-        response = self.connection.send_message(**kwargs)\n+        response = asyncio.run(self.connection.send_message(**kwargs))\n         self.log.debug(response)\n", "issue": "Support telegram-bot v20+\n### Body\n\nCurrently our telegram integration uses Telegram v13 telegram-bot library. On 1st of Jan 2023 a new, backwards incompatible version of Telegram-bot has been released : https://pypi.org/project/python-telegram-bot/20.0/#history and at least as reported by MyPy and our test suite test failures, Telegram 20 needs some changes to work:\r\n\r\nHere is a transition guide that might be helpful. \r\n\r\nTransition guide is here: https://github.com/python-telegram-bot/python-telegram-bot/wiki/Transition-guide-to-Version-20.0\r\n\r\nIn the meantime we limit telegram to < 20.0.0\n\n### Committer\n\n- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"Hook for Telegram\"\"\"\nfrom __future__ import annotations\n\nimport telegram\nimport tenacity\n\nfrom airflow.exceptions import AirflowException\nfrom airflow.hooks.base import BaseHook\n\n\nclass TelegramHook(BaseHook):\n    \"\"\"\n    This hook allows you to post messages to Telegram using the telegram python-telegram-bot library.\n\n    The library can be found here: https://github.com/python-telegram-bot/python-telegram-bot\n    It accepts both telegram bot API token directly or connection that has telegram bot API token.\n    If both supplied, token parameter will be given precedence, otherwise 'password' field in the connection\n    from telegram_conn_id will be used.\n    chat_id can also be provided in the connection using 'host' field in connection.\n    Following is the details of a telegram_connection:\n    name: 'telegram-connection-name'\n    conn_type: 'http'\n    password: 'TELEGRAM_TOKEN'\n    host: 'chat_id' (optional)\n    Examples:\n    .. code-block:: python\n\n        # Create hook\n        telegram_hook = TelegramHook(telegram_conn_id=\"telegram_default\")\n        # or telegram_hook = TelegramHook(telegram_conn_id='telegram_default', chat_id='-1xxx')\n        # or telegram_hook = TelegramHook(token='xxx:xxx', chat_id='-1xxx')\n\n        # Call method from telegram bot client\n        telegram_hook.send_message(None, {\"text\": \"message\", \"chat_id\": \"-1xxx\"})\n        # or telegram_hook.send_message(None', {\"text\": \"message\"})\n\n    :param telegram_conn_id: connection that optionally has Telegram API token in the password field\n    :param token: optional telegram API token\n    :param chat_id: optional chat_id of the telegram chat/channel/group\n    \"\"\"\n\n    def __init__(\n        self,\n        telegram_conn_id: str | None = None,\n        token: str | None = None,\n        chat_id: str | None = None,\n    ) -> None:\n        super().__init__()\n        self.token = self.__get_token(token, telegram_conn_id)\n        self.chat_id = self.__get_chat_id(chat_id, telegram_conn_id)\n        self.connection = self.get_conn()\n\n    def get_conn(self) -> telegram.bot.Bot:\n        \"\"\"\n        Returns the telegram bot client\n\n        :return: telegram bot client\n        \"\"\"\n        return telegram.bot.Bot(token=self.token)\n\n    def __get_token(self, token: str | None, telegram_conn_id: str | None) -> str:\n        \"\"\"\n        Returns the telegram API token\n\n        :param token: telegram API token\n        :param telegram_conn_id: telegram connection name\n        :return: telegram API token\n        \"\"\"\n        if token is not None:\n            return token\n\n        if telegram_conn_id is not None:\n            conn = self.get_connection(telegram_conn_id)\n\n            if not conn.password:\n                raise AirflowException(\"Missing token(password) in Telegram connection\")\n\n            return conn.password\n\n        raise AirflowException(\"Cannot get token: No valid Telegram connection supplied.\")\n\n    def __get_chat_id(self, chat_id: str | None, telegram_conn_id: str | None) -> str | None:\n        \"\"\"\n        Returns the telegram chat ID for a chat/channel/group\n\n        :param chat_id: optional chat ID\n        :param telegram_conn_id: telegram connection name\n        :return: telegram chat ID\n        \"\"\"\n        if chat_id is not None:\n            return chat_id\n\n        if telegram_conn_id is not None:\n            conn = self.get_connection(telegram_conn_id)\n            return conn.host\n\n        return None\n\n    @tenacity.retry(\n        retry=tenacity.retry_if_exception_type(telegram.error.TelegramError),\n        stop=tenacity.stop_after_attempt(5),\n        wait=tenacity.wait_fixed(1),\n    )\n    def send_message(self, api_params: dict) -> None:\n        \"\"\"\n        Sends the message to a telegram channel or chat.\n\n        :param api_params: params for telegram_instance.send_message. It can also be used to override chat_id\n        \"\"\"\n        kwargs = {\n            \"chat_id\": self.chat_id,\n            \"parse_mode\": telegram.parsemode.ParseMode.HTML,\n            \"disable_web_page_preview\": True,\n        }\n        kwargs.update(api_params)\n\n        if \"text\" not in kwargs or kwargs[\"text\"] is None:\n            raise AirflowException(\"'text' must be provided for telegram message\")\n\n        if kwargs[\"chat_id\"] is None:\n            raise AirflowException(\"'chat_id' must be provided for telegram message\")\n\n        response = self.connection.send_message(**kwargs)\n        self.log.debug(response)\n", "path": "airflow/providers/telegram/hooks/telegram.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"Hook for Telegram\"\"\"\nfrom __future__ import annotations\n\nimport asyncio\n\nimport telegram\nimport tenacity\n\nfrom airflow.exceptions import AirflowException\nfrom airflow.hooks.base import BaseHook\n\n\nclass TelegramHook(BaseHook):\n    \"\"\"\n    This hook allows you to post messages to Telegram using the telegram python-telegram-bot library.\n\n    The library can be found here: https://github.com/python-telegram-bot/python-telegram-bot\n    It accepts both telegram bot API token directly or connection that has telegram bot API token.\n    If both supplied, token parameter will be given precedence, otherwise 'password' field in the connection\n    from telegram_conn_id will be used.\n    chat_id can also be provided in the connection using 'host' field in connection.\n    Following is the details of a telegram_connection:\n    name: 'telegram-connection-name'\n    conn_type: 'http'\n    password: 'TELEGRAM_TOKEN'\n    host: 'chat_id' (optional)\n    Examples:\n    .. code-block:: python\n\n        # Create hook\n        telegram_hook = TelegramHook(telegram_conn_id=\"telegram_default\")\n        # or telegram_hook = TelegramHook(telegram_conn_id='telegram_default', chat_id='-1xxx')\n        # or telegram_hook = TelegramHook(token='xxx:xxx', chat_id='-1xxx')\n\n        # Call method from telegram bot client\n        telegram_hook.send_message(None, {\"text\": \"message\", \"chat_id\": \"-1xxx\"})\n        # or telegram_hook.send_message(None', {\"text\": \"message\"})\n\n    :param telegram_conn_id: connection that optionally has Telegram API token in the password field\n    :param token: optional telegram API token\n    :param chat_id: optional chat_id of the telegram chat/channel/group\n    \"\"\"\n\n    def __init__(\n        self,\n        telegram_conn_id: str | None = None,\n        token: str | None = None,\n        chat_id: str | None = None,\n    ) -> None:\n        super().__init__()\n        self.token = self.__get_token(token, telegram_conn_id)\n        self.chat_id = self.__get_chat_id(chat_id, telegram_conn_id)\n        self.connection = self.get_conn()\n\n    def get_conn(self) -> telegram.Bot:\n        \"\"\"\n        Returns the telegram bot client\n\n        :return: telegram bot client\n        \"\"\"\n        return telegram.Bot(self.token)\n\n    def __get_token(self, token: str | None, telegram_conn_id: str | None) -> str:\n        \"\"\"\n        Returns the telegram API token\n\n        :param token: telegram API token\n        :param telegram_conn_id: telegram connection name\n        :return: telegram API token\n        \"\"\"\n        if token is not None:\n            return token\n\n        if telegram_conn_id is not None:\n            conn = self.get_connection(telegram_conn_id)\n\n            if not conn.password:\n                raise AirflowException(\"Missing token(password) in Telegram connection\")\n\n            return conn.password\n\n        raise AirflowException(\"Cannot get token: No valid Telegram connection supplied.\")\n\n    def __get_chat_id(self, chat_id: str | None, telegram_conn_id: str | None) -> str | None:\n        \"\"\"\n        Returns the telegram chat ID for a chat/channel/group\n\n        :param chat_id: optional chat ID\n        :param telegram_conn_id: telegram connection name\n        :return: telegram chat ID\n        \"\"\"\n        if chat_id is not None:\n            return chat_id\n\n        if telegram_conn_id is not None:\n            conn = self.get_connection(telegram_conn_id)\n            return conn.host\n\n        return None\n\n    @tenacity.retry(\n        retry=tenacity.retry_if_exception_type(telegram.error.TelegramError),\n        stop=tenacity.stop_after_attempt(5),\n        wait=tenacity.wait_fixed(1),\n    )\n    def send_message(self, api_params: dict) -> None:\n        \"\"\"\n        Sends the message to a telegram channel or chat.\n\n        :param api_params: params for telegram_instance.send_message. It can also be used to override chat_id\n        \"\"\"\n        kwargs = {\n            \"chat_id\": self.chat_id,\n            \"parse_mode\": telegram.constants.ParseMode.HTML,\n            \"disable_web_page_preview\": True,\n        }\n        kwargs.update(api_params)\n\n        if \"text\" not in kwargs or kwargs[\"text\"] is None:\n            raise AirflowException(\"'text' must be provided for telegram message\")\n\n        if kwargs[\"chat_id\"] is None:\n            raise AirflowException(\"'chat_id' must be provided for telegram message\")\n\n        response = asyncio.run(self.connection.send_message(**kwargs))\n        self.log.debug(response)\n", "path": "airflow/providers/telegram/hooks/telegram.py"}]}