From f9a2e49e21832fbaeb0692b9c75c884537cfe1bf Mon Sep 17 00:00:00 2001 From: Alex Kulikov Date: Thu, 5 Feb 2026 01:40:12 +0000 Subject: [PATCH] feat: add configurable telegram api url --- src/bot.py | 12 ++++++++++-- src/tg/sender.py | 8 ++++++-- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/src/bot.py b/src/bot.py index c0ac2280..9f62be3a 100644 --- a/src/bot.py +++ b/src/bot.py @@ -75,14 +75,22 @@ def __init__( f, ) - self.application = ( + # Build Telegram application with optional custom base URL for testing + builder = ( ApplicationBuilder() .persistence(PicklePersistence(filepath="persistent_storage.pickle")) .token(tg_config["token"]) # .post_shutdown(signal_handler) .concurrent_updates(True) - .build() ) + + # Allow overriding Telegram API base URL (for mocking in tests) + telegram_base_url = os.environ.get("TELEGRAM_BASE_URL") + if telegram_base_url: + logger.info(f"Using custom Telegram base URL: {telegram_base_url}") + builder = builder.base_url(telegram_base_url) + + self.application = builder.build() self.telegram_sender = sender.TelegramSender( bot=self.application.bot, tg_config=tg_config ) diff --git a/src/tg/sender.py b/src/tg/sender.py index a9d763ae..77066a70 100644 --- a/src/tg/sender.py +++ b/src/tg/sender.py @@ -3,6 +3,7 @@ import asyncio import json import logging +import os import re import time from typing import Callable, List @@ -17,6 +18,9 @@ logger = logging.getLogger(__name__) +# Allow overriding Telegram API base URL (for mocking in tests) +TELEGRAM_API_BASE = os.environ.get("TELEGRAM_BASE_URL", "https://api.telegram.org") + class TelegramSender(Singleton): def __init__( @@ -79,7 +83,7 @@ def send_to_chat_id(self, message_text: str, chat_id: int, **kwargs) -> bool: # Handle poll sending poll_options = kwargs["poll_options"] resp = requests.get( - url=f"https://api.telegram.org/bot{self._tg_config['token']}/sendPoll", + url=f"{TELEGRAM_API_BASE}/bot{self._tg_config['token']}/sendPoll", params={ "chat_id": chat_id, "question": poll_options["question"], @@ -120,7 +124,7 @@ def send_to_chat_id(self, message_text: str, chat_id: int, **kwargs) -> bool: if "reply_markup" in kwargs: payload["reply_markup"] = kwargs["reply_markup"].to_json() resp = requests.get( - url=f"https://api.telegram.org/bot{self._tg_config['token']}/sendMessage", + url=f"{TELEGRAM_API_BASE}/bot{self._tg_config['token']}/sendMessage", json=payload, ) resp.raise_for_status()