Search code examples
pythonhttpserverbasehttpserverpython-telegram-bot

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

Solution

  • 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()
                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: after looking more closely at the code of BaseServer I found that it spins up a new instance of request handler for every request. It means that patching an instance will not work and we need to patch the class itself. Here is how it works:

    # ... 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')