Web framework adapters

RBACX ships simple adapters for popular Python web frameworks. They all follow the same conventions:

• EnvBuilder builds (Subject, Action, Resource, Context) from the framework request/scope.
• Async frameworks use async adapters (evaluate_async under the hood).
• Sync frameworks use sync adapters (evaluate_sync).
• By default, do not leak reasons. If you need diagnostics, pass add_headers=True and read:
• X-RBACX-Reason
• X-RBACX-Rule
• X-RBACX-Policy

See runnable apps in examples/.

---

FastAPI (dependency)

from fastapi import FastAPI, Depends, Request
from rbacx import Guard, Subject, Resource, Action, Context
from rbacx.adapters.fastapi import require_access


policy = {"algorithm": "deny-overrides", "rules": [
    {"id": "doc_read", "effect": "permit", "actions": ["read"], "resource": {"type": "doc"}}
]}
guard = Guard(policy)

def build_env(request: Request):
    uid = request.headers.get("x-user", "anonymous")
    return Subject(id=uid, roles=["user"]), Action("read"), Resource(type="doc"), Context()

app = FastAPI()

@app.get("/doc", dependencies=[Depends(require_access(guard, build_env, add_headers=True))])
async def doc():
    return {"ok": True}

Batch access check (UI state)

Use require_batch_access to evaluate multiple actions in one call — ideal for rendering UI elements (show/hide buttons) without N sequential requests:

from rbacx.adapters.fastapi import require_batch_access
from rbacx import Subject

def build_subject(request: Request) -> Subject:
    role = request.headers.get("X-Role", "viewer")
    return Subject(id="user", roles=[role])

@app.get("/ui-state")
async def ui_state(
    decisions=Depends(
        require_batch_access(
            guard,
            [("read", "document"), ("write", "document"), ("delete", "document")],
            build_subject,
            timeout=2.0,  # optional deadline for the whole batch
        )
    )
):
    return {
        "can_read":   decisions[0].allowed,
        "can_write":  decisions[1].allowed,
        "can_delete": decisions[2].allowed,
    }

---

Flask (decorator)

from flask import Flask, request
from rbacx import Guard, Subject, Resource, Action, Context
from rbacx.adapters.flask import require_access


guard = Guard(policy)

def build_env(req):
    # use explicit req or implicit flask.request
    r = req or request
    uid = r.headers.get("x-user", "anonymous")
    return Subject(id=uid, roles=["user"]), Action("read"), Resource(type="doc"), Context()

app = Flask(__name__)

@app.get("/doc")
@require_access(guard, build_env, add_headers=True)
def doc():
    return {"ok": True}

---

Starlette (decorator)

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from rbacx import Guard, Subject, Resource, Action, Context
from rbacx.adapters.starlette import require_access


guard = Guard(policy)

def build_env(request: Request):
    uid = request.headers.get("x-user", "anonymous")
    return Subject(id=uid, roles=["user"]), Action("read"), Resource(type="doc"), Context()

app = Starlette()

@app.route("/doc")
@require_access(guard, build_env, add_headers=True)
async def doc(request: Request):
    return JSONResponse({"ok": True})

---

Litestar (middleware)

from litestar import Litestar, get
from litestar.middleware import DefineMiddleware
from rbacx import Guard, Subject, Resource, Action, Context
from rbacx.adapters.litestar import RBACXMiddleware


guard = Guard(policy)

def build_env(scope):
    uid = dict(scope.get("headers", [])).get(b"x-user", b"anonymous").decode("latin1")
    return Subject(id=uid, roles=["user"]), Action("read"), Resource(type="doc"), Context()

@get("/doc")
async def doc() -> dict:
    return {"ok": True}

app = Litestar(
    route_handlers=[doc],
    middleware=[DefineMiddleware(RBACXMiddleware, guard=guard, build_env=build_env, add_headers=True)],
)

---

Django async (ASGI, Django 4.1+)

For Django applications running under ASGI (uvicorn, daphne), use the async variants to avoid blocking the event loop:

# settings.py
RBACX_GUARD_FACTORY = "myapp.rbacx.build_guard"
MIDDLEWARE = [
    "rbacx.adapters.django.trace.AsyncTraceIdMiddleware",
    "rbacx.adapters.django.middleware.AsyncRbacxDjangoMiddleware",
    # ...
]

from rbacx import Subject, Resource, Action, Context
from rbacx.adapters.django.decorators import async_require_access

def build_env(request):
    uid = getattr(getattr(request, "user", None), "id", None) or "anonymous"
    return Subject(id=str(uid), roles=["user"]), Action("read"), Resource(type="doc"), Context()

@async_require_access(build_env, add_headers=True)
async def doc(request):
    ...

---

Django (decorator + middleware)

Enable the middleware to inject a Guard:

# settings.py
RBACX_GUARD_FACTORY = "rbacx_demo.rbacx_factory.build_guard"
MIDDLEWARE = [
    # ...
    "rbacx.adapters.django.middleware.RbacxDjangoMiddleware",
]

Use the decorator:

from rbacx import Subject, Resource, Action, Context
from rbacx.adapters.django.decorators import require_access

def build_env(request):
    uid = getattr(getattr(request, "user", None), "id", None) or "anonymous"
    return Subject(id=str(uid), roles=["user"]), Action("read"), Resource(type="doc"), Context()

@require_access(build_env, add_headers=True)
def doc(request):
    ...

---

Django REST Framework (permission)

from rest_framework.views import APIView
from rest_framework.response import Response
from rbacx import Guard, Subject, Resource, Action, Context
from rbacx.adapters.drf import make_permission


guard = Guard(policy)

def build_env(request):
    uid = getattr(getattr(request, "user", None), "username", None) or "anonymous"
    return Subject(id=uid, roles=["user"]), Action("read"), Resource(type="doc"), Context()

RBACXPermission = make_permission(guard, build_env, add_headers=True)

# Optionally attach headers on 403:
# REST_FRAMEWORK = {"EXCEPTION_HANDLER": "rbacx.adapters.drf.rbacx_exception_handler"}

class DocsView(APIView):
    permission_classes = [RBACXPermission]
    def get(self, request):
        return Response({"ok": True})