Receiving webhooks with the SDK

import asyncio
import json
import os
from fastapi import FastAPI, Request, Response
from socialift.webhooks import parse_events, verify_challenge, verify_request

app = FastAPI()
events: asyncio.Queue = asyncio.Queue()   # in production: Redis, RQ, Celery, etc.


@app.get("/webhooks/meta")
async def meta_verify(request: Request):
    challenge = verify_challenge(dict(request.query_params), os.environ["META_VERIFY_TOKEN"])
    if challenge is None:
        return Response(status_code=403)
    return Response(challenge, media_type="text/plain")   # the bare string, not JSON


@app.post("/webhooks/meta")
async def meta_receive(request: Request):
    body = await request.body()                           # raw bytes
    if not verify_request(os.environ["FACEBOOK_APP_SECRET"], body, request.headers):
        return Response(status_code=403)
    for event in parse_events(json.loads(body)):
        events.put_nowait(event)
    return Response(status_code=200)

import json
import os
from fastapi import FastAPI, Request, Response
from socialift.webhooks import parse_tiktok_event, verify_tiktok_request

app = FastAPI()


@app.post("/webhooks/tiktok")
async def tiktok_receive(request: Request):
    body = await request.body()
    if not verify_tiktok_request(os.environ["TIKTOK_CLIENT_SECRET"], body, request.headers):
        return Response(status_code=403)
    event = parse_tiktok_event(json.loads(body))
    if event.event == "authorization.removed":
        print("Access revoked:", event.user_openid)   # delete this user's tokens
    return Response(status_code=200)

import os
from fastapi import FastAPI, Request, Response
from socialift import Telegram
from socialift.webhooks import verify_telegram_request

app = FastAPI()
SECRET = os.environ["TELEGRAM_WEBHOOK_SECRET"]    # 1–256 chars: A-Z, a-z, 0-9, _ and -


async def register_webhook():                      # once, at deploy time
    async with Telegram.from_env() as bot:
        await bot.set_webhook("https://example.uz/webhooks/telegram",
                              secret_token=SECRET, allowed_updates=["message"])


@app.post("/webhooks/telegram")
async def telegram_receive(request: Request):
    if not verify_telegram_request(request.headers, SECRET):
        return Response(status_code=403)
    update = await request.json()
    print(update.get("message", {}).get("text"))     # put it on a queue
    return Response(status_code=200)

import asyncio
from socialift.webhooks import WebhookEvent, dedupe_key

seen: set[str] = set()   # in production: a UNIQUE column in the DB or Redis SET NX


async def worker(events: asyncio.Queue):
    while True:
        event: WebhookEvent = await events.get()
        key = dedupe_key(event)                  # same as event.key
        if key not in seen:
            seen.add(key)
            if event.field == "leadgen" and event.leadgen_id:
                print("New lead:", event.leadgen_id, "form:", event.form_id)
        events.task_done()