Metadata-Version: 2.4
Name: ncheta
Version: 0.1.0
Summary: Ncheta — webhook memory. SDK for Python.
Author-email: Glintalpha <sesaalolu@gmail.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/gLiNtAlpHa/ncheta
Project-URL: Repository, https://github.com/gLiNtAlpHa/ncheta.git
Project-URL: Issues, https://github.com/gLiNtAlpHa/ncheta/issues
Keywords: webhook,relay,monitoring,replay,ncheta,webhook-memory
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: flask>=2.0; extra == "dev"
Requires-Dist: django>=3.2; extra == "dev"
Requires-Dist: fastapi>=0.68; extra == "dev"
Requires-Dist: httpx>=0.23; extra == "dev"
Requires-Dist: uvicorn>=0.17; extra == "dev"
Dynamic: license-file

# 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

```bash
pip install ncheta
```

## quickstart

### flask

```python
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

```python
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

```python
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

```python
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

```bash
# 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
