---
title: "Webhook Vault"
description: "Webhook registry API that HMAC-signs payloads with a secret from the environment, as computed columns"
url: "https://www.pixeltable.com/recipes/webhook-vault"
source: "https://github.com/pierrebrunelle/pixeltable-webhook-vault"
---

# Webhook Vault

Webhook registry API that HMAC-signs payloads with a secret from the environment, as computed columns

One `app.py` with 1 table, 3 computed columns, 4 HTTP routes. Needs `pixeltable[serve]>=0.7.14`.

## Tables

### webhooks (table)

| Column | Kind | Type or expression |
| --- | --- | --- |
| id | primary key | `pxtf.uuid.uuid7()` |
| name | stored | String |
| url | stored | String |
| event | stored | String |
| payload | stored | String |
| host | computed on insert | `target_host(url)` |
| digest | computed on insert | `digest_payload(payload)` |
| signature | computed on insert | `sign_payload(event, payload)` |

## Routes (service `vault_api`)

- `POST /webhooks`: inserts into `webhooks` (takes name, url, event, payload; returns id, host, digest, signature)
- `POST /sign`: computes `webhooks` columns without storing (takes event, payload; returns digest, signature)
- `POST /webhooks/delete`: deletes from `webhooks`
- `GET /webhooks/by-event`: runs the query `deliveries_for` (takes event)

## Run it locally

```sh
git clone https://github.com/pierrebrunelle/pixeltable-webhook-vault
cd pixeltable-webhook-vault
uv sync
uv run pxt schema update app.py vault
uv run python seed.py vault
uv run pxt service run app.py vault --port 8000
```

## Run it on Pixeltable Cloud

```sh
uv run pxt new    # a free database, no account
# put the pxt:// address it prints in pixeltable.toml
uv run pxt db update pxt://<org>:main -f
uv run pxt schema update app.py pxt://<org>:main -f
uv run pxt service update app.py pxt://<org>:main -f
```

## app.py

At commit `d5c41554` (https://github.com/pierrebrunelle/pixeltable-webhook-vault/blob/d5c415548de09477b9a765ad6df7acb6e8434a00/app.py):

```python
"""Webhook Vault API built with Pixeltable.

    export WEBHOOK_HMAC_SECRET=<any-long-random-string>
    pxt schema update app.py vault
    pxt service run app.py vault
"""
import pixeltable as pxt
import pixeltable.functions as pxtf
from pixeltable.serving import FastAPIRouter

from udfs import digest_payload, sign_payload, target_host

# ---- tables ----
TableModel = pxt.model_base()


class Webhooks(TableModel, name='webhooks'):
    id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
    name: pxt.String
    url: pxt.String
    event: pxt.String
    payload: pxt.String

    host = target_host(url)
    digest = digest_payload(payload)
    signature = sign_payload(event, payload)


# ---- queries ----
@pxt.query
def deliveries_for(event: str):
    """Signed deliveries for one event type."""
    return Webhooks.where(Webhooks.event == event).select(
        Webhooks.id, Webhooks.name, Webhooks.host, Webhooks.signature
    ).order_by(Webhooks.name)


# ---- routes ----
vault_api = FastAPIRouter(name='vault_api')
vault_api.add_insert_route(
    Webhooks, path='/webhooks',
    inputs=[Webhooks.name, Webhooks.url, Webhooks.event, Webhooks.payload],
    outputs=[Webhooks.id, Webhooks.host, Webhooks.digest, Webhooks.signature],
)
vault_api.add_compute_route(Webhooks, path='/sign', inputs=[Webhooks.event, Webhooks.payload],
                            outputs=[Webhooks.digest, Webhooks.signature])
vault_api.add_delete_route(Webhooks, path='/webhooks/delete')
vault_api.add_query_route(path='/webhooks/by-event', query=deliveries_for, method='get')
```