Skip to main content
Use Kernite as a synchronous guard before every write endpoint.

​
Route Pattern

from __future__ import annotations

from fastapi import FastAPI, HTTPException, Request

from kernite import evaluate_execute

app = FastAPI()


def build_execute_request(req: Request, payload: dict) -> dict:
    principal_id = req.headers.get("x-principal-id", "api:unknown")
    return {
        "workspace_id": payload["workspace_id"],
        "principal": {
            "type": "token",
            "id": principal_id,
            "attributes": {"role": payload.get("role", "member")},
        },
        "object_type": "invoice",
        "operation": "create",
        "payload": payload,
        "policy_context": {
            "governed": True,
            "selected_policies": [
                {
                    "policy_key": "invoice_create_guard",
                    "policy_version": 1,
                    "effect": "allow",
                    "rules": [
                        {
                            "rule_key": "require_currency",
                            "rule_definition": {
                                "type": "required_fields",
                                "fields": ["currency"],
                            },
                            "reason_code": "required_currency",
                            "reason_message": "currency is required.",
                        }
                    ],
                }
            ],
        },
    }


@app.post("/invoices")
async def create_invoice(req: Request, payload: dict):
    execute_request = build_execute_request(req, payload)
    governance = evaluate_execute(
        execute_request,
        idempotency_key=req.headers.get("idempotency-key"),
    )

    if governance["data"]["decision"] != "approved":
        raise HTTPException(
            status_code=403,
            detail={
                "reason_codes": governance["data"]["reason_codes"],
                "governance": governance,
            },
        )

    created_invoice = {"id": "inv-001", **payload}
    return {"ok": True, "data": created_invoice, "governance": governance}

​
Integration Rules

  • Evaluate in the same request path right before mutation.
  • Branch on data.decision, not message.
  • Persist ctx_id, data.trace_hash, and data.idempotency_key with the write.
  • Do not trust a cached allow flag from previous requests.