Litestar app with DI, role guards and exception handlers: working versions (tested 2026-10-15)

litestar==2.24.0
pytest==9.1.1

from litestar import Litestar, Request, delete, get
from litestar.connection import ASGIConnection
from litestar.di import Provide
from litestar.exceptions import NotAuthorizedException, PermissionDeniedException
from litestar.handlers import BaseRouteHandler
from litestar.response import Response


class ItemNotFound(Exception):
    pass


class ItemRepo:
    def __init__(self, items: dict[int, str] | None = None):
        self.items = items if items is not None else {1: "widget"}

    def get(self, item_id: int) -> str:
        if item_id not in self.items:
            raise ItemNotFound(item_id)
        return self.items[item_id]

    def delete(self, item_id: int) -> None:
        if self.items.pop(item_id, None) is None:
            raise ItemNotFound(item_id)


async def provide_repo() -> ItemRepo:
    # swap for a real database session in production
    return ItemRepo()


def role_guard(connection: ASGIConnection, handler: BaseRouteHandler) -> None:
    # demo only: trusts a header; use a verified token in production
    required = handler.opt.get("role")
    if required is None:
        return
    role = connection.headers.get("x-role")
    if role is None:
        raise NotAuthorizedException("missing credentials")
    if role != required:
        raise PermissionDeniedException(f"requires role {required}")


@get("/items/{item_id:int}")
async def get_item(item_id: int, repo: ItemRepo) -> dict:
    return {"id": item_id, "name": repo.get(item_id)}


@delete("/items/{item_id:int}", opt={"role": "admin"})
async def delete_item(item_id: int, repo: ItemRepo) -> None:
    repo.delete(item_id)


def item_not_found(request: Request, exc: ItemNotFound) -> Response:
    return Response({"error": "item not found", "id": exc.args[0]}, status_code=404)


ROUTES = [get_item, delete_item]
GUARDS = [role_guard]
HANDLERS = {ItemNotFound: item_not_found}

app = Litestar(
    route_handlers=ROUTES,
    dependencies={"repo": Provide(provide_repo)},
    guards=GUARDS,
    exception_handlers=HANDLERS,
)

import pytest
from litestar.di import Provide
from litestar.testing import create_test_client

from app import GUARDS, HANDLERS, ROUTES, ItemRepo


@pytest.fixture
def repo():
    return ItemRepo({1: "widget", 2: "gadget"})


@pytest.fixture
def client(repo):
    async def provide() -> ItemRepo:
        # the in-memory fake overrides the production provider
        return repo

    with create_test_client(
        route_handlers=ROUTES,
        dependencies={"repo": Provide(provide)},
        guards=GUARDS,
        exception_handlers=HANDLERS,
    ) as c:
        yield c


def test_get_returns_item_without_role(client):
    r = client.get("/items/1")
    assert r.status_code == 200
    assert r.json() == {"id": 1, "name": "widget"}


def test_missing_item_uses_exception_handler(client):
    r = client.get("/items/99")
    assert r.status_code == 404
    assert r.json() == {"error": "item not found", "id": 99}


def test_delete_without_credentials_is_401(client, repo):
    r = client.delete("/items/1")
    assert r.status_code == 401
    assert 1 in repo.items


def test_delete_with_wrong_role_is_403(client, repo):
    r = client.delete("/items/1", headers={"x-role": "user"})
    assert r.status_code == 403
    assert 1 in repo.items


def test_admin_delete_removes_item(client, repo):
    r = client.delete("/items/1", headers={"x-role": "admin"})
    assert r.status_code == 204
    assert 1 not in repo.items

[pytest]
pythonpath = .