Msgspec Integration

This recipe illustrates how the popular msgspec data serialization and validation library can be used with Falcon.

Media handlers

msgspec can be used for JSON serialization by simply instantiating JSONHandler with the msgspec.json.decode() and msgspec.json.encode() functions:

import msgspec

import falcon.media

json_handler = falcon.media.JSONHandler(
    dumps=msgspec.json.encode,
    loads=msgspec.json.decode,
)

Note

It would be more efficient to use preconstructed msgspec.json.Decoder and msgspec.json.Encoder instead of initializing them every time via msgspec.json.decode() and msgspec.json.encode(), respectively. This optimization is left as an exercise for the reader.

Attention

As seen above, plugging msgspec into JSONHandler may look extremely straightforward. We have, however, omitted an important caveat when it comes to error handling – see a more detailed discussion in the below chapter: Error handling.

Using msgspec for handling MessagePack (or YAML) media is slightly more involved, as we need to implement the BaseHandler interface:

from typing import Optional

from msgspec import msgpack

from falcon import media
from falcon.typing import ReadableIO


class MsgspecMessagePackHandler(media.BaseHandler):
    def deserialize(
        self,
        stream: ReadableIO,
        content_type: Optional[str],
        content_length: Optional[int],
    ) -> object:
        return msgpack.decode(stream.read())

    def serialize(self, media: object, content_type: str) -> bytes:
        return msgpack.encode(media)


msgpack_handler = MsgspecMessagePackHandler()

We can now use these handlers for request and response media (see also: Replacing the Default Handlers).

Media validation

Falcon currently only provides optional media validation using JSON Schema, so we will implement validation separately using process_resource middleware. To that end, let us assume that resources can expose schema attributes based on the HTTP verb: PATCH_SCHEMA, POST_SCHEMA, etc, pointing to the msgspec.Struct in question. We will inject the validated object into params:

import msgspec

from falcon import Request
from falcon import Response


class MsgspecMiddleware:
    def process_resource(
        self, req: Request, resp: Response, resource: object, params: dict
    ) -> None:
        if schema := getattr(resource, f'{req.method}_SCHEMA', None):
            param = schema.__name__.lower()
            params[param] = msgspec.convert(req.get_media(), schema)

Here, the name of the injected parameter is simply a lowercase version of the schema’s class name. We could instead store this name in a constant on the resource, or on the schema class.

Note that while the above middleware is only suitable for synchronous (WSGI) Falcon applications, you could easily transform this method to async, or even implement an async version in parallel on the same middleware as process_request_async(...). Just do not forget to await req.get_media()!

The astute reader may also notice that the process can be optimized even further, as msgspec affords combined deserialization and validation in a single call to decode() without going via an intermediate Python object (in this case, effectively req.media). However, if one wants to support more than just a single request media handler, all the media functionality would need to be reimplemented almost from scratch.

We are looking into providing better affordances for this scenario in the framework itself (see also #2202 on GitHub), as well as offering centralized media validation connected to the application’s OpenAPI specification. Your input is extremely valuable for us, so do not hesitate to get in touch, and share your vision!

Error handling

Schema validation can fail, and the resulting exception would unexpectedly bubble up as a generic HTTP 500 error. We can do better! Skimming through msgspec's docs, we find out that this case is represented by msgspec.ValidationError. We could either create an error hander that reraises this exception as MediaValidationError, or just use a try.. except clause, and reraise directly inside middleware.

Our customized JSONHandler has another issue: unlike the stdlib’s json or the majority of other JSON libraries, msgspec.DecodeError is not a subclass of ValueError:

>>> import msgspec
>>> issubclass(msgspec.DecodeError, ValueError)
False

This discrepancy can be worked around using a wrapper around msgspec.json.decode() that reraises a ValueError from msgspec.DecodeError. Furthermore, the problem has been reported upstream, and received positive feedback from the maintainer, so hopefully it could get resolved in the near future.

Complete recipe

Finally, we combine these snippets into a note taking application:

from datetime import datetime
from datetime import timezone
from functools import partial
from http import HTTPStatus
from typing import Annotated, Any
import uuid

import msgspec

import falcon
from falcon import Request
from falcon import Response
from falcon.media import JSONHandler


def _utc_now() -> datetime:
    return datetime.now(timezone.utc)


class Note(msgspec.Struct):
    text: Annotated[str, msgspec.Meta(max_length=256)]
    noteid: uuid.UUID = msgspec.field(default_factory=uuid.uuid4)
    created: datetime = msgspec.field(
        default_factory=partial(datetime.now, timezone.utc)
    )


class NoteResource:
    POST_SCHEMA = Note

    def __init__(self) -> None:
        # NOTE: In a real-world app, you would want to use persistent storage.
        self._store = {}

    def on_get_note(self, req: Request, resp: Response, noteid: uuid.UUID) -> None:
        resp.media = self._store.get(str(noteid))
        if not resp.media:
            raise falcon.HTTPNotFound(
                description=f'Note with {noteid=} is not in the store.'
            )

    def on_delete_note(self, req: Request, resp: Response, noteid: uuid.UUID) -> None:
        self._store.pop(str(noteid), None)
        resp.status = HTTPStatus.NO_CONTENT

    def on_get(self, req: Request, resp: Response) -> None:
        resp.media = self._store

    def on_post(self, req: Request, resp: Response, note: Note) -> None:
        self._store[str(note.noteid)] = note
        resp.location = f'{req.path}/{note.noteid}'
        resp.media = note
        resp.status = HTTPStatus.CREATED


class MsgspecMiddleware:
    def process_resource(
        self, req: Request, resp: Response, resource: object, params: dict[str, Any]
    ) -> None:
        if schema := getattr(resource, f'{req.method}_SCHEMA', None):
            param = schema.__name__.lower()
            params[param] = msgspec.convert(req.get_media(), schema)


def _msgspec_loads(content: str) -> Any:
    try:
        return msgspec.json.decode(content)
    except msgspec.DecodeError as ex:
        raise falcon.MediaMalformedError(falcon.MEDIA_JSON) from ex


def _handle_validation_error(
    req: Request, resp: Response, ex: msgspec.ValidationError, params: dict[str, Any]
) -> None:
    raise falcon.HTTPUnprocessableEntity(description=str(ex))


def create_app() -> falcon.App:
    app = falcon.App(middleware=[MsgspecMiddleware()])
    app.add_error_handler(msgspec.ValidationError, _handle_validation_error)

    json_handler = JSONHandler(
        dumps=msgspec.json.encode,
        loads=_msgspec_loads,
    )
    app.req_options.media_handlers[falcon.MEDIA_JSON] = json_handler
    app.resp_options.media_handlers[falcon.MEDIA_JSON] = json_handler

    notes = NoteResource()
    app.add_route('/notes', notes)
    app.add_route('/notes/{noteid:uuid}', notes, suffix='note')

    return app


application = create_app()

We can now POST a new note, view the whole collection, or just GET an individual note. (MessagePack support was omitted for brevity.)