ncheta 0.1.0

Ncheta — webhook memory. SDK for Python.

ncheta

your webhooks just got memory. Python SDK.

ncheta (igbo: memory / remembrance) is a self-hostable webhook monitoring tool. it intercepts incoming webhooks, stores the raw request before your handler runs, then forwards it. if your handler crashes — the data is safe. fix the bug. replay. done.

no more "stripe sent it once and we missed it."

install

pip install ncheta

quickstart

flask

from flask import Flask, request, jsonify
from ncheta import Ncheta

app = Flask(__name__)

ncheta = Ncheta(control_port=7001, ingestion_port=7000)
ncheta.start()

ncheta.client.create_endpoint(
    "stripe", "http://127.0.0.1:5000/webhooks/stripe"
)

watch = ncheta.watch_flask({
    "endpoint_name": "stripe",
    "target_url": "http://127.0.0.1:5000/webhooks/stripe",
})

@app.route("/webhooks/stripe", methods=["POST"])
@watch
def stripe_webhook():
    payload = request.get_json(silent=True)
    return jsonify({"received": True})

fastapi

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from ncheta import Ncheta

ncheta = Ncheta(control_port=7001, ingestion_port=7000)

@asynccontextmanager
async def lifespan(app: FastAPI):
    ncheta.start()
    ncheta.client.create_endpoint(
        "stripe", "http://127.0.0.1:8000/webhooks/stripe"
    )
    yield
    ncheta.stop()

app = FastAPI(lifespan=lifespan)

middleware_fn = ncheta.watch_fastapi({
    "endpoint_name": "stripe",
    "target_url": "http://127.0.0.1:8000/webhooks/stripe",
})
app.middleware("http")(middleware_fn)

@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
    body = await request.json()
    return JSONResponse({"received": True})

django

from ncheta import Ncheta

ncheta = Ncheta(control_port=7001, ingestion_port=7000)
ncheta.start()

NchetaMiddleware = ncheta.watch_django({
    "endpoint_name": "stripe",
    "target_url": "http://127.0.0.1:8000/webhooks/stripe",
})

# add to MIDDLEWARE in settings.py
MIDDLEWARE = [
    # ...
    "yourapp.ncheta_setup.NchetaMiddleware",
]

how it works

[Stripe] ──POST──> [ncheta :7000] ──stores──> [SQLite]
                        │
                        ├── returns 200 to Stripe immediately
                        │
                        └── forwards to your handler in background
                                │
                                ├── handler returns 200? done.
                                └── handler returns 500? retry with backoff.

api

ncheta = Ncheta(
    control_port=7001,   # control API port
    ingestion_port=7000, # where webhooks land
    db_url=None,         # defaults to sqlite://./ncheta.db
    start_timeout=10.0,  # seconds to wait for binary
)

ncheta.start()                              # start the binary
ncheta.stop()                               # stop the binary

ncheta.client.health()                      # GET /health
ncheta.client.endpoints()                   # GET /endpoints
ncheta.client.create_endpoint(name, url)    # POST /endpoints
ncheta.client.events()                      # GET /events
ncheta.client.event(event_id)               # GET /events/:id
ncheta.client.attempts(event_id)            # GET /events/:id/attempts
ncheta.client.replay(event_id)              # POST /events/:id/replay

replay

# event failed? fix your handler, then:
curl -X POST http://localhost:7001/events/{id}/replay

# or from python:
ncheta.replay(event_id)

config

all env vars. all optional. sane defaults.

var	default	what it does
NCHETA_DB_URL	sqlite://./ncheta.db	database connection
NCHETA_INGESTION_PORT	7000	where webhooks land
NCHETA_CONTROL_PORT	7001	REST API + dashboard
NCHETA_MAX_BODY_SIZE	524288 (512KB)	max request/response body
NCHETA_MAX_RETRIES	5	retries before giving up
NCHETA_EVENT_TTL_SECONDS	604800 (7 days)	auto-cleanup after this

requirements

• Python 3.8+
• Zero runtime dependencies (uses only stdlib urllib)
• Framework middleware requires: flask, django, or fastapi (as applicable)

license

MIT