diff --git a/telebot/__init__.py b/telebot/__init__.py index 2ac86cf06..a43621ca3 100644 --- a/telebot/__init__.py +++ b/telebot/__init__.py @@ -414,10 +414,18 @@ def load_reply_handlers(self, filename="./.handler-saves/reply.save", del_file_a self.reply_backend.load_handlers(filename, del_file_after_loading) - def set_webhook(self, url: Optional[str]=None, certificate: Optional[Union[str, Any]]=None, max_connections: Optional[int]=None, - allowed_updates: Optional[List[str]]=None, ip_address: Optional[str]=None, - drop_pending_updates: Optional[bool] = None, timeout: Optional[int]=None, secret_token: Optional[str]=None) -> bool: - """ + def set_webhook( + self, + url: Optional[str] = None, + certificate: Optional[Union[str, BinaryIO]] = None, + max_connections: Optional[int] = None, + allowed_updates: Optional[List[str]] = None, + ip_address: Optional[str] = None, + drop_pending_updates: Optional[bool] = None, + timeout: Optional[int] = None +) -> bool: + + """ Use this method to specify a URL and receive incoming updates via an outgoing webhook. Whenever there is an update for the bot, we will send an HTTPS POST request to the specified URL, containing a JSON-serialized Update. In case of an unsuccessful request, we will give up after @@ -466,12 +474,10 @@ def set_webhook(self, url: Optional[str]=None, certificate: Optional[Union[str, :return: True on success. :rtype: :obj:`bool` if the request was successful. """ - return apihelper.set_webhook( - self.token, url = url, certificate = certificate, max_connections = max_connections, - allowed_updates = allowed_updates, ip_address = ip_address, drop_pending_updates = drop_pending_updates, - timeout = timeout, secret_token = secret_token) - + self.token, url, certificate, max_connections, + allowed_updates, ip_address, drop_pending_updates, timeout + ) def run_webhooks(self, listen: Optional[str]="127.0.0.1", diff --git a/telebot/apihelper.py b/telebot/apihelper.py index 879dfdec5..590f81b66 100644 --- a/telebot/apihelper.py +++ b/telebot/apihelper.py @@ -21,11 +21,11 @@ import telebot from telebot import types from telebot import util +from telebot.exceptions import TelebotAPIError, NetworkError logger = telebot.logger proxy = None -session = None API_URL = None FILE_URL = None @@ -35,7 +35,7 @@ LONG_POLLING_TIMEOUT = 10 # Should be positive, short polling should be used for testing purposes only (https://core.telegram.org/bots/api#getupdates) -SESSION_TIME_TO_LIVE = 600 # In seconds. None - live forever, 0 - one-time +_session = requests.Session() # Global session for connection pooling (created on RETRY_ON_ERROR = False RETRY_TIMEOUT = 2 @@ -169,6 +169,57 @@ def _make_request(token, method_name, method='get', params=None, files=None): if json_result: return json_result['result'] +def send_request( + token: str, + method_name: str, + params: Optional[Dict[str, Any]] = None, + files: Optional[Dict[str, Any]] = None, + method: str = 'post', + timeout: Optional[int] = None +) -> Any: + request_url = _get_request_url(token, method_name) + headers = {'User-Agent': 'pyTelegramBotAPI'} + + try: + if files: + # Use multipart/form-data for file uploads + response = _session.post( + request_url, + data=params, + files=files, + timeout=timeout, + headers=headers + ) + elif method.lower() == 'post': + # Use application/json for non-file POST requests + response = _session.post( + request_url, + json=params, + timeout=timeout, + headers=headers + ) + else: + # GET parameters are sent in the URL + response = _session.get( + request_url, + params=params, + timeout=timeout, + headers=headers + ) + # Handle response (existing logic) + # ... + except requests.exceptions.RequestException as e: + logger.error("Request failed: %s", e) + raise NetworkError(f"Network error: {str(e)}") from e + + # Check API response for errors + result = _parse(response) + if isinstance(result, dict) and not result.get('ok', False): + raise TelebotAPIError( + description=result.get('description', 'Unknown error'), + error_code=result.get('error_code') + ) + return result # Re-raise for custom exception handling def _check_result(method_name, result): """ @@ -274,30 +325,31 @@ def send_message( payload['allow_paid_broadcast'] = allow_paid_broadcast return _make_request(token, method_url, params=payload, method='post') - -def set_webhook(token, url=None, certificate=None, max_connections=None, allowed_updates=None, ip_address=None, - drop_pending_updates = None, timeout=None, secret_token=None): +def set_webhook( + token: str, + url: Optional[str] = None, + certificate: Optional[Union[str, BinaryIO]] = None, + max_connections: Optional[int] = None, + allowed_updates: Optional[List[str]] = None, + ip_address: Optional[str] = None, + drop_pending_updates: Optional[bool] = None, + timeout: Optional[int] = None +) -> bool: method_url = r'setWebhook' - payload = { - 'url': url if url else "", - } - files = None - if certificate: - files = {'certificate': certificate} - if max_connections: - payload['max_connections'] = max_connections - if allowed_updates is not None: # Empty lists should pass - payload['allowed_updates'] = json.dumps(allowed_updates) - if ip_address is not None: # Empty string should pass - payload['ip_address'] = ip_address - if drop_pending_updates is not None: # Any bool value should pass - payload['drop_pending_updates'] = drop_pending_updates - if timeout: - payload['timeout'] = timeout - if secret_token: - payload['secret_token'] = secret_token - return _make_request(token, method_url, params=payload, files=files) - + params: Dict[str, Any] = {'url': url or ''} + files = {'certificate': certificate} if certificate else None + + if max_connections is not None: + params['max_connections'] = max_connections + if allowed_updates: + params['allowed_updates'] = json.dumps(allowed_updates) + if ip_address: + params['ip_address'] = ip_address + if drop_pending_updates is not None: + params['drop_pending_updates'] = drop_pending_updates + + result = _make_request(token, method_url, params, files=files, method='post', timeout=timeout) + return result if isinstance(result, bool) else False def delete_webhook(token, drop_pending_updates=None, timeout=None): method_url = r'deleteWebhook' diff --git a/telebot/exceptions.py b/telebot/exceptions.py new file mode 100644 index 000000000..e3dcdb9d8 --- /dev/null +++ b/telebot/exceptions.py @@ -0,0 +1,14 @@ +class TelebotException(Exception): + """Base exception for all library exceptions.""" + pass + +class TelebotAPIError(TelebotException): + """Raised when Telegram API returns an error.""" + def __init__(self, description: str, error_code: Optional[int] = None): + self.description = description + self.error_code = error_code + super().__init__(f"Error {error_code}: {description}" if error_code else description) + +class NetworkError(TelebotException): + """Raised for network-related errors (e.g., timeouts).""" + pass