Python quickstart

mkdir npayload-fastapi && cd npayload-fastapi
python -m venv .venv
source .venv/bin/activate
pip install npayload fastapi uvicorn

NPAYLOAD_CLIENT_ID=oac_your_client_id
NPAYLOAD_HMAC_SECRET=your_hmac_secret
WEBHOOK_SECRET=your_webhook_secret

# npayload_client.py
import os
from npayload import NPayloadAuth, NPayloadClient

auth = NPayloadAuth(
    client_id=os.environ["NPAYLOAD_CLIENT_ID"],
    hmac_secret=os.environ["NPAYLOAD_HMAC_SECRET"],
)

npayload = NPayloadClient(auth=auth, environment="development")

# setup.py
from npayload_client import npayload

async def setup():
    # Create a channel
    await npayload.channels.create(
        name="orders",
        description="Order lifecycle events",
        privacy="standard",
    )

    # Subscribe via webhook
    await npayload.subscriptions.create(
        channel="orders",
        name="order-processor",
        type="webhook",
        endpoint={
            "url": f"{os.environ.get('BASE_URL', 'http://localhost:8000')}/webhooks/orders",
            "method": "POST",
        },
    )

    print("Channel and subscription created.")

# main.py
import os
from contextlib import asynccontextmanager
from dotenv import load_dotenv

load_dotenv()

from fastapi import FastAPI, Request, HTTPException
from npayload_client import npayload
from setup import setup


@asynccontextmanager
async def lifespan(app: FastAPI):
    await setup()
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/publish")
async def publish(request: Request):
    body = await request.json()
    message = await npayload.messages.publish(
        channel="orders",
        payload=body,
    )
    return {"messageId": message.gid}


@app.post("/webhooks/orders")
async def webhook_handler(request: Request):
    body = await request.json()
    signature = request.headers.get("x-npayload-signature", "")

    is_valid = npayload.webhooks.verify(
        body,
        signature,
        os.environ["WEBHOOK_SECRET"],
    )

    if not is_valid:
        raise HTTPException(status_code=401, detail="Invalid signature")

    print(f"Received: {body.get('payload')}")
    return {"received": True}

uvicorn main:app --reload --port 8000

curl -X POST http://localhost:8000/publish \
  -H "Content-Type: application/json" \
  -d '{"event": "order.created", "orderId": "ord_123", "total": 59.98}'