Добавьте пользовательское сопоставление запросов во встроенный HTTPServer.HTTPServer в python-telegram-bot

Question

Я только что переписал мой бот телеграммы из pyTelegramBotAPI с python-telegram-bot, Была идея сделать общедоступный URL-адрес мониторинга, который мы могли бы время от времени пинговать с некоторым приложением, чтобы увидеть, работает ли бот (внутренне, он будет тестировать некоторые функции бота, такие как база данных и т. Д.). Вопрос в том, можно ли создать такую ​​точку запроса, используя встроенную HTTPServer.HTTPServer? До сих пор я не мог найти способ сделать это. Если я повторно использую общее example.com/botTOKEN метод, мне нужно позаботиться о полезной нагрузке json, и я не могу отправить обратно HTTP-коды ошибок в случае сбоя.

Спасибо.

Обновление 1: Итак, я следовал фрагментам кода, предоставленным @Marat. Вот как я получаю объект-обработчик:

# since the webhook server is started in an extra thread, it's not available immediately
while updater.httpd is None:
    pass
handler = updater.httpd.RequestHandlerClass

Answer 1

Да, ты можешь. Я надеюсь, что этот пример поможет:

import SimpleHTTPServer
import SocketServer


class myServer(SimpleHTTPServer.SimpleHTTPRequestHandler):

    def do_GET(self):
        """Serve a GET request."""
        # do something here
        self.finish(
            "Hello world! Request path: " + self.path
        )

    def finish(self, value, status=200, ctype="text/html"):
        try:
            self.send_response(status)
            self.send_header("Content-type", ctype)
            self.send_header("Content-Length", str(len(value)))
            self.end_headers()
            self.wfile.write(str(value))
        finally:
            self.wfile.close()


httpd = SocketServer.TCPServer(("", 80), myServer)

httpd.serve_forever()

Для получения дополнительной информации, посмотрите на исходный код SimpleHTTPRequestHandler

UPD: код WebhookHandler может быть лучшим примером

Если вы хотите повторно использовать существующий экземпляр, вы можете сделать патч обезьяны:

# wrapper for original do_GET
def patch(get_func):
    def wrapper(self):
        if self.path == '/test_url':
            # do something
            message = "Successful" # or not :(
            self.send_response(200)
            self.send_header("Content-type", 'text/plain')
            self.send_header("Content-Length", str(len(message)))
            self.end_headers()
            self.wfile.write(message)
        else:
            return get_func(self)

    return wrapper

# assume `server` is an instance of WebhookHandler
server.do_GET = patch(server.do_GET)  # monkeypatching

UPD2: после более внимательного изучения кода BaseServer я обнаружил, что он запускает новый экземпляр обработчика запросов для каждого запроса. Это означает, что исправление экземпляра не будет работать, и нам нужно исправить сам класс. Вот как это работает:

# ... somewhere in the code far far away

from telegram.ext import Updater
from telegram.utils import webhookhandler as wh

# IMPORTANT: do it before making an instance of updater
# IMPORTANT #2: this is considered dirty hack. Don't repeat it at home!
if not wh.WebhookHandler._fuse:
    # note we're patching handler class itself, not an instance
    wh.WebhookHandler.do_GET = patch(wh.WebhookHandler.do_GET)  # use patch() from above
    wh.WebhookHandler._fuse = True

updater = Updater(token='TOKEN')