Как ограничить количество подключений прокси в Pyrogram?

Question

[Alex_888]

в Telethon есть такое:
connection_retries=int(),
request_retries=int(),
retry_delay=int(),
auto_reconnect=True,
timeout=int(),

и если прокси невалид, то подключение ограничивается выставленным мной количеством попыток.

в Pyrogram документации не нашел подобного. Он до талого пытается подключиться! Как его ограничить можно?

Answer 1

[Михаил Р.]

в Pyrogram документации не нашел подобного

Все верно, Pyrogram не настолько гибкий по прокси, в сравнение с Telethon. Используйте внешний менеджер подключений/проксей и не будите зависеть от конкретной библиотеки, например:

import asyncio
from pyrogram import Client
from pyrogram.errors import PyrogramError

class ProxyManager:
    def __init__(self, app, max_retries=5, retry_delay=1, proxy=None):
        self.app = app
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        self.proxy = proxy
        self.client = None

    async def connect(self):
        for attempt in range(1, self.max_retries + 1):
            try:
                print(f"Connection attempt #{attempt}")
                self.client = Client(**self.app, proxy=self.proxy)
                await self.client.start()
                print("The connection was established successfully.")

            except PyrogramError as e:
                print(f"Connection error: {e}")
                if attempt == self.max_retries:
                    print("The maximum number of connection attempts has been reached, stop!")
                    break

                await asyncio.sleep(self.retry_delay)

    async def disconnect(self):
        if self.client:
            await self.client.stop()
            print("Connection is closed.")

app = {
    'api_id': 'YOUR_API_ID',
    'api_hash': 'YOUR_API_HASH',
    'session_name': 'your_session_name'
}

proxy = {
    'scheme': 'http',  # or 'socks5'
    'hostname': 'your.proxy.hostname',
    'port': 1080,
    'username': 'user',
    'password': 'password'
}

# test run
async def main():
    manager = ProxyManager(app, max_retries=3, retry_delay=2, proxy=proxy)
    await manager.connect()
    await manager.disconnect()

asyncio.run(main())

Comments

[Alex_888]

Казалось бы это довольно важная вещь, печально что разрабы так к ней отнеслись.
Спасибо!