TelegramWalletPay

1
import os
2

3
from aiohttp import web
4
from telegram_wallet_pay.schemas import WebhookMessages
5
from telegram_wallet_pay.tools.aiohttp import check_signature
6

7
# store TELEGRAM_WALLET_PAY_TOKEN to your .env
8
# wallet token can be issued via https://pay.wallet.tg/
9
TELEGRAM_WALLET_PAY_TOKEN = os.getenv("TELEGRAM_WALLET_PAY_TOKEN")
10

11

12
# add `check_signature` decorator to your handler
13
@check_signature(TELEGRAM_WALLET_PAY_TOKEN)
14
async def webhook_handler(request: web.Request) -> web.StreamResponse:
15
    """Handle webhook from Telegram Wallet API."""
16
    json_data = await request.text()
17
    webhook_messages = WebhookMessages.model_validate_json(json_data)
18
    for webhook_message in webhook_messages:
19
        # process every webhook message as you wish
20
        # e.g. store them to your database. We just print it here
21
        print(f"Received webhook message: {webhook_message}")
22

23
    return web.json_response({"success": True})
24

25

26
if __name__ == "__main__":
27
    app = web.Application()
28
    routes = web.RouteTableDef()
29
    web.run_app(app)
30