Webhook¶

You get a WebhookApp from Bot.create_webhook_app(). The guide is Webhooks.

class WebhookApp( bot, *, secret, path=None, url=None, max_body=DEFAULT_MAX_BODY, ack_timeout=8.0, allow_no_secret=False, )¶

The ASGI application the server sends updates to.

Get one from Bot.create_webhook_app. It is a complete ASGI 3 application: it answers HTTP requests and takes the lifespan events, so that, served on its own, it starts the bot when the web server starts and stops it at shutdown. WebSocket connections are closed.

Each request is checked in this order, and the status code tells the server what to do:

404: a different path, when path was given;
405, with Allow: POST: not a POST;
401, with WWW-Authenticate: the secret header is missing or wrong. It is compared in constant time, before the body is read;
413: the body is larger than max_body;
408: the body did not arrive within 10 seconds;
200: the update is safe (stored, or handled with MemoryStorage), or it is a repeat. Also a request with the right secret whose body is not an update: it is logged as an error, because a 4xx would make the server send it again for days, and every later update would wait behind it;
503: the update is not safe yet (it could not be stored, or the handlers took longer than ack_timeout with MemoryStorage), or the bot is stopping. The server tries again later;
500: an unexpected error, logged. The server tries again later.

Inside another application, the app gets no lifespan events of its own: pass lifespan=bot.lifespan to that application. Without it, the bot starts at the first delivery and logs a warning. For a web framework that is not ASGI, handle does the checks and the processing.

url¶

The public address the bot registers with the server at every start, or None.

Type:: str | None

Examples

On its own, served with uvicorn bot:app:

app = bot.create_webhook_app(
    secret=os.environ["WEBHOOK_SECRET"],
    url="https://bot.example.org/pgpchat",
)

In FastAPI, next to your own routes:

from fastapi import FastAPI

webhook = bot.create_webhook_app(
    secret=os.environ["WEBHOOK_SECRET"], url="https://example.org/pgpchat"
)
app = FastAPI(lifespan=bot.lifespan)
app.add_route("/pgpchat", webhook, methods=["POST"])

In Starlette, with the same webhook:

from starlette.applications import Starlette
from starlette.routing import Route

app = Starlette(
    routes=[Route("/pgpchat", webhook, methods=["POST"])],
    lifespan=bot.lifespan,
)

Warning

Give the app a route, not a mount. With the app mounted at /pgpchat, Starlette answers a POST to /pgpchat with a redirect to /pgpchat/, and the server counts a redirect as a failed delivery: the queue stops. If you do mount it, register the address with the trailing slash.

Sets up the endpoint. Nothing is sent to the server here.

Use Bot.create_webhook_app rather than this: it passes the bot, and Bot.lifespan then knows which url to register.

Parameters:

bot (Bot | SyncBot) – The bot that runs the updates.
secret (str | None) – The secret the server sends in the X-PGPChat-Bot-Api-Secret-Token header, checked on every delivery: 1 to 256 characters of A-Z, a-z, 0-9, _ and -. generate_secret makes a good one. None only with allow_no_secret=True.
path (str | None) – Answer only on this path, and 404 elsewhere. A trailing / makes no difference. Leave it out when another application routes to the app.
url (str | None) – The app’s public https address. When given, the bot registers it with the server, with the secret, every time it starts.
max_body (int) – The largest body accepted, in bytes: 1 MiB by default. A larger one gets 413.
ack_timeout (float) – With MemoryStorage, how many seconds to wait for the handlers before answering 503. Above 0 and under 10, the server’s own time limit.
allow_no_secret (bool) – Accept secret=None.

Raises:

ConfigurationError – secret is None without allow_no_secret=True, or ack_timeout is not above 0 and under 10.
ValidationError – secret or url is malformed.

Warning

With allow_no_secret=True and no secret, anyone who finds the address can send the bot fake updates. Use it only when something in front of the app already checks that requests come from the server.

property secret: str | None¶: The secret checked on every delivery, or None if the app was made without one.

async handle(headers, body, *, peer=None)¶

Checks and processes one delivery, for a web framework that is not ASGI.

It checks the secret, the size and the shape, then hands the update to the bot and answers as the ASGI app does. The path, the method and the 10-second limit for reading the body are left to your framework. Start the bot before the first delivery (await bot.start()) and stop it at the end (await bot.stop()).

Parameters:

headers (Mapping[str, str]) – The request headers. Names are matched without regard to case.
body (bytes) – The raw request body.
peer (str | None) – The client’s address, only used in log messages.

Returns:

The status, body and headers to answer with. Anything but a 2xx makes the server send the update again later.

Return type:

WebhookResult

Example

A route in aiohttp:

from aiohttp import web


async def pgpchat(request: web.Request) -> web.Response:
    body = await request.read()
    result = await webhook.handle(request.headers, body, peer=request.remote)
    return web.json_response(
        result.body, status=result.status, headers=dict(result.headers)
    )

class WebhookResult(status, body, headers=())[source]¶

The answer handle chose, to send back to the server.

A 2xx status tells the server the update was delivered. Anything else makes it send the same update again later, and the updates after it wait.

status¶

The HTTP status code to answer with.

Type:: int

body¶

The body, to send as JSON: {"ok": True} with a 2xx, {"ok": False} otherwise.

Type:: dict[str, Any]

headers¶

Extra response headers, as (name, value) pairs with lower-case names. Empty except on a 401, which carries www-authenticate.

Type:: tuple[tuple[str, str], …]

generate_secret(nbytes=32)¶

Makes a random secret for a webhook, in the alphabet the server accepts.

Pass it to Bot.create_webhook_app. Make it once and keep it with your other secrets: the server sends the secret it was given with each delivery, so every worker process must use that same one.

Parameters:: nbytes (int) – How many random bytes. The default, 32, gives 43 characters. The server takes 1 to 256 characters, so keep it between 1 and 192.
Returns:: The secret, made of letters, digits, _ and - by secrets.token_urlsafe.
Return type:: str

Examples

The default length:

>>> len(generate_secret())
43

To make one in a terminal:

python -c "from pgpcbot import generate_secret; print(generate_secret())"

SECRET_HEADER = 'x-pgpchat-bot-api-secret-token'¶

The HTTP header that carries the webhook secret, in lower case.

The server sends it as X-PGPChat-Bot-Api-Secret-Token with every delivery, when the webhook has a secret. WebhookApp compares header names without regard to case.