fastapi echo template

[Quiago]

I try the code for the fastpi webhook example: #!/usr/bin/env python

-- coding: utf-8 --

This is a simple echo bot using decorators and webhook with fastapi

It echoes any incoming text messages and does not use the polling method.

import logging
import fastapi
import uvicorn
import telebot

API_TOKEN = 'TOKEN'

WEBHOOK_HOST = '<ip/domain>'
WEBHOOK_PORT = 8443 # 443, 80, 88 or 8443 (port need to be 'open')
WEBHOOK_LISTEN = '0.0.0.0' # In some VPS you may need to put here the IP addr

WEBHOOK_SSL_CERT = './webhook_cert.pem' # Path to the ssl certificate
WEBHOOK_SSL_PRIV = './webhook_pkey.pem' # Path to the ssl private key

Quick'n'dirty SSL certificate generation:

openssl genrsa -out webhook_pkey.pem 2048

openssl req -new -x509 -days 3650 -key webhook_pkey.pem -out webhook_cert.pem

When asked for "Common Name (e.g. server FQDN or YOUR name)" you should reply

with the same value in you put in WEBHOOK_HOST

WEBHOOK_URL_BASE = "https://{}:{}".format(WEBHOOK_HOST, WEBHOOK_PORT)
WEBHOOK_URL_PATH = "/{}/".format(API_TOKEN)

logger = telebot.logger
telebot.logger.setLevel(logging.INFO)

bot = telebot.TeleBot(API_TOKEN)

app = fastapi.FastAPI(docs=None, redoc_url=None)

@app.post(f'/{API_TOKEN}/')
def process_webhook(update: dict):
"""
Process webhook calls
"""
if update:
update = telebot.types.Update.de_json(update)
bot.process_new_updates([update])
else:
return

@bot.message_handler(commands=['help', 'start'])
def send_welcome(message):
"""
Handle '/start' and '/help'
"""
bot.reply_to(message,
("Hi there, I am EchoBot.\n"
"I am here to echo your kind words back to you."))

@bot.message_handler(func=lambda message: True, content_types=['text'])
def echo_message(message):
"""
Handle all other messages
"""
bot.reply_to(message, message.text)

Remove webhook, it fails sometimes the set if there is a previous webhook

bot.remove_webhook()

Set webhook

bot.set_webhook(
url=WEBHOOK_URL_BASE + WEBHOOK_URL_PATH,
certificate=open(WEBHOOK_SSL_CERT, 'r')
)

uvicorn.run(
app,
host=WEBHOOK_LISTEN,
port=WEBHOOK_PORT,
ssl_certfile=WEBHOOK_SSL_CERT,
ssl_keyfile=WEBHOOK_SSL_PRIV
)
But I have this error: Traceback (most recent call last):
File "/opt/render/project/src/.venv/bin/uvicorn", line 8, in
sys.exit(main())
^^^^^^
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/click/core.py", line 1161, in call
return self.main(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/click/core.py", line 1082, in main
rv = self.invoke(ctx)
^^^^^^^^^^^^^^^^
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/click/core.py", line 1443, in invoke
return ctx.invoke(self.callback, **ctx.params)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/click/core.py", line 788, in invoke
return __callback(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/uvicorn/main.py", line 412, in main
run(
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/uvicorn/main.py", line 579, in run
server.run()
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/uvicorn/server.py", line 66, in run
return asyncio.run(self.serve(sockets=sockets))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/render/project/python/Python-3.11.11/lib/python3.11/asyncio/runners.py", line 190, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/opt/render/project/python/Python-3.11.11/lib/python3.11/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/render/project/python/Python-3.11.11/lib/python3.11/asyncio/base_events.py", line 654, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/uvicorn/server.py", line 70, in serve
await self._serve(sockets)
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/uvicorn/server.py", line 77, in _serve
config.load()
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/uvicorn/config.py", line 435, in load
self.loaded_app = import_from_string(self.app)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/uvicorn/importer.py", line 19, in import_from_string
module = importlib.import_module(module_str)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/render/project/python/Python-3.11.11/lib/python3.11/importlib/init.py", line 126, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "", line 1204, in _gcd_import
File "", line 1176, in _find_and_load
File "", line 1147, in _find_and_load_unlocked
File "", line 690, in _load_unlocked
File "", line 940, in exec_module
File "", line 241, in _call_with_frames_removed
File "/opt/render/project/src/main.py", line 84, in
uvicorn.run(
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/uvicorn/main.py", line 579, in run
server.run()
File "/opt/render/project/src/.venv/lib/python3.11/site-packages/uvicorn/server.py", line 66, in run
return asyncio.run(self.serve(sockets=sockets))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/opt/render/project/python/Python-3.11.11/lib/python3.11/asyncio/runners.py", line 186, in run
raise RuntimeError(
RuntimeError: asyncio.run() cannot be called from a running event loop
sys:1: RuntimeWarning: coroutine 'Server.serve' was never awaited

[loci456]

this code should work fine.

import telebot
import logging
from fastapi.responses import JSONResponse
import ssl
import fastapi
import uvicorn


API_TOKEN = '1111111:AAAAAAAAAAAAAAAAAAAAAAAA'
WEBHOOK_HOST = '1.1.1.1'
WEBHOOK_PORT = 443  # 443, 80, 88 or 8443
WEBHOOK_LISTEN = '0.0.0.0'

WEBHOOK_SSL_CERT = '/webhook_cert.pem'
WEBHOOK_SSL_PRIV = '/webhook_pkey.pem'

WEBHOOK_URL_BASE = "https://{}:{}".format(WEBHOOK_HOST, WEBHOOK_PORT)
WEBHOOK_URL_PATH = "/{}/".format(API_TOKEN)

logger = telebot.logger
telebot.logger.setLevel(logging.INFO)
bot = telebot.TeleBot(API_TOKEN)

app = fastapi.FastAPI(docs_url=None, redoc_url=None)


@app.get("/docs", include_in_schema=False)
async def docs_disable():
    return JSONResponse(
        status_code=404, content={'message': "Not found"}, )


@app.get("/redoc", include_in_schema=False)
async def redoc_disable():
    return JSONResponse(
        status_code=404, content={'message': "Not found"}, )


@app.post(f'/{API_TOKEN}/')
async def process_webhook(update: dict):
    if update:
        update = telebot.types.Update.de_json(update)
        bot.process_new_updates([update])
    else:
        return


@app.get("/{full_path:path}")
async def capture_routes():
    return JSONResponse(
        status_code=404, content={'message': "Not found"},)


@bot.message_handler(commands=['help', 'start'])
def send_welcome(message):
    bot.reply_to(message,
                 ("Hi there, I am EchoBot.\n"
                  "I am here to echo your kind words back to you."))


@bot.message_handler(func=lambda message: True, content_types=['text'])
def echo_message(message):
    bot.reply_to(message, message.text)


context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain(WEBHOOK_SSL_CERT, WEBHOOK_SSL_PRIV)
uvicorn.run(
    app,
    host=WEBHOOK_LISTEN,
    port=WEBHOOK_PORT,
    ssl_certfile=WEBHOOK_SSL_CERT,
    ssl_keyfile=WEBHOOK_SSL_PRIV
)

you can install the certificate like this:

curl https://api.telegram.org/bot1111111:AAAAAAAAAAAAAAAAAAAAAAAA/setWebhook -F "url=https://1.1.1.1:443/1111111:AAAAAAAAAAAAAAAAAAAAAAAA/" -F "certificate=@webhook_cert.pem"

also make sure that the port you selected is open.