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