AuthzX + FastAPI

Add authorization to your FastAPI app in 5 minutes.

Install

pip install authzx

Set up the client

import os
from authzx import AuthzX

authzx = AuthzX(api_key=os.environ["AUTHZX_API_KEY"])

Built-in dependency

The SDK includes a FastAPI dependency out of the box:

from fastapi import FastAPI, Depends

app = FastAPI()

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

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

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

The require() dependency extracts subject ID from the X-User-ID header by default. Returns 401 if missing, 403 if denied.

Custom subject header

If your user ID comes from a different header:

@app.get("/documents/{id}")
async def get_document(
    id: str,
    _=Depends(authzx.require("document", "read", subject_header="authorization-user-id"))
):
    return {"id": id}

Manual check

For more control, call async_check() directly:

from fastapi import FastAPI, Request, HTTPException
from authzx 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 authzx.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 authzx import AuthorizeRequest, Action

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

    resp = await authzx.async_authorize(AuthorizeRequest(
        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

authzx = AuthzX(base_url="http://localhost:8181")

No API key needed. Same API, sub-millisecond responses.

Client lifecycle

The SDK reuses HTTP connections. Close on shutdown:

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

Error handling

from authzx import AuthzXError

@app.exception_handler(AuthzXError)
async def authzx_error_handler(request, exc):
    if exc.is_auth_error:
        return JSONResponse(status_code=500, content={"error": "AuthzX configuration error"})
    return JSONResponse(status_code=502, content={"error": "Authorization service error"})