Support Local Bot API Server (with custom `api_url`).

davtelepot/__init__.py

@@ -11,7 +11,7 @@ __author__ = "Davide Testa"
 __email__ = "davide@davte.it"
 __credits__ = ["Marco Origlia", "Nick Lee @Nickoala"]
 __license__ = "GNU General Public License v3.0"
-__version__ = "2.9.9"
+__version__ = "2.9.10"
 __maintainer__ = "Davide Testa"
 __contact__ = "t.me/davte"
 

davtelepot/api.py

@@ -358,6 +358,7 @@ class TelegramBot:
     All mirrored methods are camelCase.
     """
     _loop = None
+    _api_url = "https://api.telegram.org/bot"
 
     app = aiohttp.web.Application()
     sessions_timeouts = {
@@ -374,12 +375,13 @@ class TelegramBot:
     _per_chat_cooldown_timedelta = datetime.timedelta(seconds=1)
     _allowed_messages_per_group_per_minute = 20
 
-    def __init__(self, token):
+    def __init__(self, token, api_url: str = None):
         """Set bot token and store HTTP sessions."""
         if self.loop is None:
             self.__class__._loop = asyncio.new_event_loop()
             asyncio.set_event_loop(self.loop)
         self._token = token
+        self._api_url = api_url
         self.sessions = dict()
         self._flood_wait = 0
         # Each `telegram_id` key has a list of `datetime.datetime` as value
@@ -399,6 +401,18 @@ class TelegramBot:
         """Telegram API bot token."""
         return self._token
 
+    @property
+    def api_url(self):
+        """Telegram API bot token."""
+        return self._api_url or self.__class__._api_url
+
+    @classmethod
+    def set_class_api_url(cls, api_url: str):
+        cls._api_url = api_url
+
+    def set_api_url(self, api_url: str):
+        self._api_url = api_url
+
     @property
     def flood_wait(self):
         """Seconds to wait before next API requests."""
@@ -627,7 +641,7 @@ class TelegramBot:
             await self.prevent_flooding(parameters['chat_id'])
         parameters = self.adapt_parameters(parameters, exclude=exclude)
         try:
-            async with session.post("https://api.telegram.org/bot"
+            async with session.post(f"{self.api_url}/bot"
                                     f"{self.token}/{method}",
                                     data=parameters) as response:
                 try:

davtelepot/bot.py

@@ -104,10 +104,10 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
     _errors_file_path = None
     _documents_max_dimension = 50 * 1000 * 1000  # 50 MB
 
-    def __init__(
-        self, token, hostname='', certificate=None, max_connections=40,
-        allowed_updates=None, database_url='bot.db'
-    ):
+    def __init__(self,
+                 token, hostname='', certificate=None,
+                 max_connections=40, allowed_updates=None,
+                 database_url='bot.db', api_url: str = None):
         """Init a bot instance.
 
         token : str
@@ -125,7 +125,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
         # Append `self` to class list of instances
         self.__class__.bots.append(self)
         # Call superclasses constructors with proper arguments
-        TelegramBot.__init__(self, token)
+        TelegramBot.__init__(self, token, api_url=api_url)
         ObjectWithDatabase.__init__(self, database_url=database_url)
         MultiLanguageObject.__init__(self)
         self.messages['davtelepot'] = davtelepot_messages
@@ -2140,7 +2140,7 @@ class Bot(TelegramBot, ObjectWithDatabase, MultiLanguageObject):
             return file
         file_bytes = await async_get(
             url=(
-                f"https://api.telegram.org/file/"
+                f"{self.api_url}/file/"
                 f"bot{self.token}/"
                 f"{file['file_path']}"
             ),