d56 d56 - 4 days ago 5
Python Question

Add custom request mapping in python-telegram-bot embedded HTTPServer.HTTPServer

i have just rewritten my telegram bot from

pyTelegramBotAPI
with
python-telegram-bot
. There was an idea to have a monitoring url available publicly that we could ping once in a while with some app to see if the bot is still running (internally, it would test some functionality of the bot, like the database, etc). The question is, whether it is possible to create such a request point using the embedded
HTTPServer.HTTPServer
? So far, i couldn't find a way to do that. If i reuse the general
example.com/botTOKEN
method i need to take care for a json payload and i cannot send back HTTP error response codes in case of failure.

Thank you.

Update1: So, I followed the code snippets provided by @Marat. This is how i am getting hold of the handler object:

# 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

Yes, you can. I hope this example will help:

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()

For more information, look at the source code of SimpleHTTPRequestHandler

UPD: WebhookHandler code could be a better example

If you want to reuse the existing instance, you can do monkey patching:

# 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()
        else:
            return get_func(self)

    return wrapper

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