"""

_request = None

def __init__(self,

token=None,

base_url=None,

workers=4,

bot=None,

user_sig_handler=None,

request_kwargs=None):

if (token is None) and (bot is None):

raise ValueError('`token` or `bot` must be passed')

if (token is not None) and (bot is not None):

raise ValueError('`token` and `bot` are mutually exclusive')

self.logger = logging.getLogger(__name__)

con_pool_size = workers + 4

if bot is not None:

self.bot = bot

if bot.request.con_pool_size < con_pool_size:

self.logger.warning(

'Connection pool of Request object is smaller than optimal value (%s)',

con_pool_size)

else:

# we need a connection pool the size of:

# * for each of the workers

# * 1 for Dispatcher

# * 1 for polling Updater (even if webhook is used, we can spare a connection)

# * 1 for JobQueue

# * 1 for main thread

if request_kwargs is None:

request_kwargs = {}

if 'con_pool_size' not in request_kwargs:

request_kwargs['con_pool_size'] = con_pool_size

self._request = Request(**request_kwargs)

self.bot = Bot(token, base_url, request=self._request)

self.user_sig_handler = user_sig_handler

self.update_queue = Queue()

self.job_queue = JobQueue(self.bot)

self.__exception_event = Event()

self.dispatcher = Dispatcher(

self.bot,

self.update_queue,

job_queue=self.job_queue,

workers=workers,

exception_event=self.__exception_event)

self.last_update_id = 0

self.running = False

self.is_idle = False

self.httpd = None

self.__lock = Lock()

self.__threads = []

def _init_thread(self, target, name, *args, **kwargs):

thr = Thread(target=self._thread_wrapper, name=name, args=(target,) + args, kwargs=kwargs)

thr.start()

self.__threads.append(thr)

def _thread_wrapper(self, target, *args, **kwargs):

thr_name = current_thread().name

self.logger.debug('{0} - started'.format(thr_name))

try:

target(*args, **kwargs)

except Exception:

self.__exception_event.set()

self.logger.exception('unhandled exception')

raise

self.logger.debug('{0} - ended'.format(thr_name))

def start_polling(self,

poll_interval=0.0,

timeout=10,

network_delay=None,

clean=False,

bootstrap_retries=0,

read_latency=2.,

allowed_updates=None):

"""Starts polling updates from Telegram.

if network_delay is not None:

warnings.warn('network_delay is deprecated, use read_latency instead')

read_latency = network_delay

with self.__lock:

if not self.running:

self.running = True

# Create & start threads

self.job_queue.start()

dispatcher_ready = Event()

self._init_thread(self.dispatcher.start, "dispatcher", ready=dispatcher_ready)

self._init_thread(self._start_polling, "updater", poll_interval, timeout,

read_latency, bootstrap_retries, clean, allowed_updates)

dispatcher_ready.wait()

# Return the update queue so the main thread can insert updates

return self.update_queue

def start_webhook(self,

listen='127.0.0.1',

port=80,

url_path='',

cert=None,

key=None,

clean=False,

bootstrap_retries=0,

webhook_url=None,

allowed_updates=None):

"""

with self.__lock:

if not self.running:

self.running = True

# Create & start threads

self.job_queue.start()

self._init_thread(self.dispatcher.start, "dispatcher"),

self._init_thread(self._start_webhook, "updater", listen, port, url_path, cert,

key, bootstrap_retries, clean, webhook_url, allowed_updates)

# Return the update queue so the main thread can insert updates

return self.update_queue

def _start_polling(self, poll_interval, timeout, read_latency, bootstrap_retries, clean,

allowed_updates): # pragma: no cover

# """

# Thread target of thread 'updater'. Runs in background, pulls

# updates from Telegram and inserts them in the update queue of the

# Dispatcher.

# """

cur_interval = poll_interval

self.logger.debug('Updater thread started')

self._bootstrap(bootstrap_retries, clean=clean, webhook_url='', allowed_updates=None)

while self.running:

try:

updates = self.bot.get_updates(

self.last_update_id,

timeout=timeout,

read_latency=read_latency,

allowed_updates=allowed_updates)

except RetryAfter as e:

self.logger.info(str(e))

cur_interval = 0.5 + e.retry_after

except TelegramError as te:

self.logger.error("Error while getting Updates: {0}".format(te))

# Put the error into the update queue and let the Dispatcher

# broadcast it

self.update_queue.put(te)

cur_interval = self._increase_poll_interval(cur_interval)

else:

if not self.running:

if len(updates) > 0:

self.logger.debug('Updates ignored and will be pulled '

'again on restart.')

break

if updates:

for update in updates:

self.update_queue.put(update)

self.last_update_id = updates[-1].update_id + 1

cur_interval = poll_interval

sleep(cur_interval)

@staticmethod

def _increase_poll_interval(current_interval):

# increase waiting times on subsequent errors up to 30secs

if current_interval == 0:

current_interval = 1

elif current_interval < 30:

current_interval += current_interval / 2

elif current_interval > 30:

current_interval = 30

return current_interval

def _start_webhook(self, listen, port, url_path, cert, key, bootstrap_retries, clean,

webhook_url, allowed_updates):

self.logger.debug('Updater thread started')

use_ssl = cert is not None and key is not None

if not url_path.startswith('/'):

url_path = '/{0}'.format(url_path)

# Create and start server

self.httpd = WebhookServer((listen, port), WebhookHandler, self.update_queue, url_path,

self.bot)

if use_ssl:

self._check_ssl_cert(cert, key)

# DO NOT CHANGE: Only set webhook if SSL is handled by library

if not webhook_url:

webhook_url = self._gen_webhook_url(listen, port, url_path)

self._bootstrap(

max_retries=bootstrap_retries,

clean=clean,

webhook_url=webhook_url,

cert=open(cert, 'rb'),

allowed_updates=allowed_updates)

elif clean:

self.logger.warning("cleaning updates is not supported if "

"SSL-termination happens elsewhere; skipping")

self.httpd.serve_forever(poll_interval=1)

def _check_ssl_cert(self, cert, key):

# Check SSL-Certificate with openssl, if possible

try:

exit_code = subprocess.call(

["openssl", "x509", "-text", "-noout", "-in", cert],

stdout=open(os.devnull, 'wb'),

stderr=subprocess.STDOUT)

except OSError:

exit_code = 0

if exit_code is 0:

try:

self.httpd.socket = ssl.wrap_socket(

self.httpd.socket, certfile=cert, keyfile=key, server_side=True)

except ssl.SSLError as error:

self.logger.exception('Failed to init SSL socket')

raise TelegramError(str(error))

else:

raise TelegramError('SSL Certificate invalid')

@staticmethod

def _gen_webhook_url(listen, port, url_path):

return 'https://{listen}:{port}{path}'.format(listen=listen, port=port, path=url_path)

def _bootstrap(self, max_retries, clean, webhook_url, allowed_updates, cert=None):

retries = 0

while 1:

try:

if clean:

# Disable webhook for cleaning

self.bot.delete_webhook()

self._clean_updates()

sleep(1)

self.bot.set_webhook(

url=webhook_url, certificate=cert, allowed_updates=allowed_updates)

except (Unauthorized, InvalidToken):

raise

except TelegramError:

msg = 'error in bootstrap phase; try={0} max_retries={1}'.format(retries,

max_retries)

if max_retries < 0 or retries < max_retries:

self.logger.warning(msg)

retries += 1

else:

self.logger.exception(msg)

raise

else:

break

sleep(1)

def _clean_updates(self):

self.logger.debug('Cleaning updates from Telegram server')

updates = self.bot.get_updates()

while updates:

updates = self.bot.get_updates(updates[-1].update_id + 1)

def stop(self):

"""Stops the polling/webhook thread, the dispatcher and the job queue."""

self.job_queue.stop()

with self.__lock:

if self.running or self.dispatcher.has_running_threads:

self.logger.debug('Stopping Updater and Dispatcher...')

self.running = False

self._stop_httpd()

self._stop_dispatcher()

self._join_threads()

# Stop the Request instance only if it was created by the Updater

if self._request:

self._request.stop()

def _stop_httpd(self):

if self.httpd:

self.logger.debug('Waiting for current webhook connection to be '

'closed... Send a Telegram message to the bot to exit '

'immediately.')

self.httpd.shutdown()

self.httpd = None

def _stop_dispatcher(self):

self.logger.debug('Requesting Dispatcher to stop...')

self.dispatcher.stop()

def _join_threads(self):

for thr in self.__threads:

self.logger.debug('Waiting for {0} thread to end'.format(thr.name))

thr.join()

self.logger.debug('{0} thread has ended'.format(thr.name))

self.__threads = []

def signal_handler(self, signum, frame):

self.is_idle = False

if self.running:

self.stop()

if self.user_sig_handler:

self.user_sig_handler(signum, frame)

else:

self.logger.warning('Exiting immediately!')

import os

os._exit(1)

def idle(self, stop_signals=(SIGINT, SIGTERM, SIGABRT)):

"""Blocks until one of the signals are received and stops the updater.

for sig in stop_signals:

signal(sig, self.signal_handler)

self.is_idle = True

while self.is_idle: