Python

pip install requests python-dotenv
# Optional for FastAPI apps:
# pip install fastapi uvicorn

# File: test_ingest.py  ← SMOKE TEST ONLY
# Run once:  python test_ingest.py
# This does NOT keep listening for future crashes.
import os
import requests
from dotenv import load_dotenv

load_dotenv()

url = "https://lawde.net/api/v1/logs/ingest"
headers = {
    "x-lawde-api-key": os.environ["LAWDE_API_KEY"],
    "Content-Type": "application/json",
}
payload = {
    "error_class": "ValueError",
    "error_message": "LAWDE Shield Initial Test Crash",
    "file_path": "test_ingest.py",
    "line_number": 1,
    "stack_trace": "ValueError: LAWDE Shield Initial Test Crash\n  at test_ingest.py:1",
    "environment": "development",
    "app_id": os.environ.get("LAWDE_APP_ID"),
}
payload = {k: v for k, v in payload.items() if v is not None}
r = requests.post(url, json=payload, headers=headers, timeout=15)
print(r.status_code, r.json())

# File: app/lawde_report.py  ← LIVE helper used by your framework
import os
import traceback
import requests
from dotenv import load_dotenv

load_dotenv()

INGEST_URL = "https://lawde.net/api/v1/logs/ingest"
API_KEY = os.environ["LAWDE_API_KEY"]
APP_ID = os.environ.get("LAWDE_APP_ID")  # optional


def report_exception(exc: BaseException, *, environment: str = "production") -> dict | None:
    """Push one crash to Sentinel. Never raise into the host app."""
    tb = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
    frame = traceback.extract_tb(exc.__traceback__)[-1] if exc.__traceback__ else None
    payload = {
        "error_class": type(exc).__name__,
        "error_message": str(exc) or type(exc).__name__,
        "stack_trace": tb or str(exc),
        "file_path": frame.filename if frame else "unknown",
        "line_number": frame.lineno if frame else 1,
        "environment": environment,
    }
    if APP_ID:
        payload["app_id"] = APP_ID

    try:
        response = requests.post(
            INGEST_URL,
            headers={
                "Content-Type": "application/json",
                "x-lawde-api-key": API_KEY,
            },
            json=payload,
            timeout=15,
        )
        response.raise_for_status()
        return response.json()
    except Exception:
        return None  # fail silently — never take down the API

# File: app/main.py  ← LIVE wire for FastAPI
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from app.lawde_report import report_exception

app = FastAPI()


@app.exception_handler(Exception)
async def lawde_exception_handler(request: Request, exc: Exception):
    # Called for unhandled exceptions in routes — this IS live monitoring
    report_exception(exc, environment="production")
    return JSONResponse(status_code=500, content={"detail": "internal_error"})


@app.get("/boom")
def boom():
    raise RuntimeError("LAWDE live wire check")


# Start (example):
# uvicorn app.main:app --host 0.0.0.0 --port 8000

# File: app.py  ← LIVE wire for Flask
from flask import Flask, jsonify
from app.lawde_report import report_exception

app = Flask(__name__)


@app.errorhandler(Exception)
def handle_exception(exc):
    report_exception(exc, environment="production")
    return jsonify(error="internal_error"), 500


@app.get("/boom")
def boom():
    raise RuntimeError("LAWDE live wire check")


if __name__ == "__main__":
    app.run(port=5000)

# File: myproject/lawde_middleware.py  ← LIVE wire for Django
from app.lawde_report import report_exception  # adjust import path


class LawdeExceptionMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        return self.get_response(request)

    def process_exception(self, request, exception):
        # Django calls this for unhandled view exceptions
        report_exception(exception, environment="production")
        return None  # let Django continue its normal 500 handling


# settings.py — add near the top of MIDDLEWARE so it wraps views:
# MIDDLEWARE = [
#     "myproject.lawde_middleware.LawdeExceptionMiddleware",
#     ...
# ]

# Celery / RQ / custom workers — LIVE wire
# Call report_exception inside your task failure hooks or a process-wide sys.excepthook.

import sys
from app.lawde_report import report_exception

def _lawde_excepthook(exc_type, exc, tb):
    try:
        if exc is not None:
            report_exception(exc, environment="production")
    finally:
        sys.__excepthook__(exc_type, exc, tb)

sys.excepthook = _lawde_excepthook
# Import this module at worker boot (celery worker_process_init, etc.)