Skip to main content

Vengtoo + FastAPI

Add authorization to your FastAPI app in 5 minutes.

Install​

pip install vengtoo

Set up the client​

import os
from vengtoo import Vengtoo

vengtoo = Vengtoo(api_key=os.environ["VENGTOO_API_KEY"])

Built-in dependency​

The SDK includes a FastAPI dependency out of the box:

from fastapi import FastAPI, Depends, Request
from vengtoo import Subject

app = FastAPI()

# Resolve the authenticated caller from the request (set by your auth layer).
def current_subject(request: Request) -> Subject:
    return Subject(id=request.state.user_id, type="user")

@app.get("/documents/{id}")
async def get_document(id: str, _=Depends(vengtoo.require("document", "read", current_subject))):
    return {"id": id, "content": "..."}

@app.put("/documents/{id}")
async def update_document(id: str, _=Depends(vengtoo.require("document", "write", current_subject))):
    return {"updated": True}

@app.delete("/documents/{id}")
async def delete_document(id: str, _=Depends(vengtoo.require("document", "delete", current_subject))):
    return {"deleted": True}

The require() dependency calls your current_subject resolver to identify the caller. Return a Subject with no id (or raise) to reject with 401; a denied decision returns 403.

Custom subject resolution​

current_subject is where you adapt to your auth layer — a different header, a decoded JWT, or session state:

def current_subject(request: Request) -> Subject:
    return Subject(external_id=request.headers["authorization-user-id"], type="user")

Manual check​

For more control, call async_check() directly:

from fastapi import FastAPI, Request, HTTPException
from vengtoo import Subject, Resource

@app.get("/documents/{id}")
async def get_document(id: str, request: Request):
    user_id = request.headers.get("x-user-id")
    if not user_id:
        raise HTTPException(status_code=401, detail="unauthorized")

    allowed = await vengtoo.async_check(
        subject=Subject(id=user_id, type="user"),
        action="read",
        resource=Resource(id=id, type="document"),
    )

    if not allowed:
        raise HTTPException(status_code=403, detail="forbidden")

    return {"id": id, "content": "..."}

Full response with reason​

from vengtoo import EvaluationRequest, Action

@app.get("/documents/{id}")
async def get_document(id: str, request: Request):
    user_id = request.headers.get("x-user-id")

    resp = await vengtoo.async_evaluate(EvaluationRequest(
        subject=Subject(id=user_id, type="user"),
        resource=Resource(id=id, type="document"),
        action=Action(name="read"),
    ))

    if not resp.decision:
        raise HTTPException(status_code=403, detail=resp.context.reason)

    return {"id": id, "policy": resp.context.policy_id, "path": resp.context.access_path}

Using the local agent​

vengtoo = Vengtoo(base_url="http://localhost:8181")

No API key needed. Same API, evaluated locally, with no round-trip to the cloud.

Client lifecycle​

The SDK reuses HTTP connections. Close on shutdown:

@app.on_event("shutdown")
async def shutdown():
    await vengtoo.async_close()

Error handling​

from vengtoo import VengtooError

@app.exception_handler(VengtooError)
async def vengtoo_error_handler(request, exc):
    if exc.is_auth_error:
        return JSONResponse(status_code=500, content={"error": "Vengtoo configuration error"})
    return JSONResponse(status_code=502, content={"error": "Authorization service error"})
Last updated on