Complete example¶

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

from msgspec import Struct

from jero import BaseApp, BaseFactory, HTTPError, JSONResponse, Resource


class InvalidTokenError(
    HTTPError,
    type="invalid-token",
    title="Invalid token",
    status=401,
): ...


class WidgetNotFoundError(
    HTTPError,
    type="widget-not-found",
    title="Widget not found",
    status=404,
): ...


class Credentials(Struct):
    authorization: str


class User(Struct):
    id: str
    name: str


class WidgetPath(Struct):
    widget_id: str


class WidgetIn(Struct):
    name: str


class Widget(WidgetIn):
    id: str
    owner_id: str


class WidgetHeaders(Struct, omit_defaults=True):
    x_trace_id: str | None = None


class TokenAuth:
    def authenticate(self, headers: Credentials) -> User:
        token = headers.authorization.removeprefix("Bearer ").strip()
        if token != "token":
            raise InvalidTokenError()
        return User(id="user-id", name="user-name")


@dataclass
class WidgetStore:
    _widgets: dict[str, Widget]

    async def get(self, widget_id: str) -> Widget:
        try:
            return self._widgets[widget_id]
        except KeyError:
            raise WidgetNotFoundError() from None

    async def list_for_user(self, user: User) -> list[Widget]:
        return [widget for widget in self._widgets.values() if widget.owner_id == user.id]

    async def create(self, user: User, widget: WidgetIn) -> Widget:
        widget_id = f"widget-{len(self._widgets) + 1}"
        created = Widget(id=widget_id, owner_id=user.id, name=widget.name)
        self._widgets[created.id] = created
        return created


@dataclass
class WidgetService:
    _store: WidgetStore

    async def get_widget(self, user: User, widget_id: str) -> Widget:
        widget = await self._store.get(widget_id)
        if widget.owner_id != user.id:
            raise WidgetNotFoundError()
        return widget

    async def list_widgets(self, user: User) -> list[Widget]:
        return await self._store.list_for_user(user)

    async def create_widget(self, user: User, widget: WidgetIn) -> Widget:
        return await self._store.create(user, widget)


@asynccontextmanager
async def open_widget_store() -> AsyncIterator[WidgetStore]:
    widgets = {
        "widget-1": Widget(id="widget-1", owner_id="user-id", name="first-widget"),
    }
    yield WidgetStore(widgets)


class Factory(BaseFactory):
    async def create_widget_service(self) -> WidgetService:
        store = await self._aenter(open_widget_store())
        return WidgetService(store)


@dataclass
class WidgetResource(Resource, path="/widgets"):
    _service: WidgetService

    async def create(
        self,
        json: WidgetIn,
        user: User,
    ) -> JSONResponse[Widget, WidgetHeaders]:
        widget = await self._service.create_widget(user, json)
        return JSONResponse(json=widget, headers=WidgetHeaders(x_trace_id="trace-id"))

    async def read_many(self, user: User) -> list[Widget]:
        return await self._service.list_widgets(user)

    async def read_one(self, path: WidgetPath, user: User) -> Widget:
        return await self._service.get_widget(user, path.widget_id)


class App(BaseApp[Factory]):
    async def wire(self) -> None:
        auth = TokenAuth()
        widget_service = await self._factory.create_widget_service()
        self._include_resource(WidgetResource(widget_service), auth=auth)


app = App()

granian --interface asgi myapp:app

curl -H "Authorization: Bearer token" localhost:8000/widgets/widget-1
# {"name":"first-widget","id":"widget-1","ownerId":"user-id"}