JSON encoding¶
EDRI supports custom JSON encoding for REST and WebSocket payloads.
This document describes how to implement a custom encoder, how encoder context works, and how to serialize complex types.
Where the encoder is used¶
A custom encoder is used when EDRI renders JSON for:
REST responses (
application/json)WebSocket messages (JSON payloads)
Configuring a custom encoder¶
Pass your encoder class when starting the API:
edri.start_api(
middlewares=[...],
json_encoder=MyJSONEncoder,
)
Base class: CustomJSONEncoder¶
Custom encoders should inherit from:
edri.utility.json_encoder.CustomJSONEncoder
Override default(...) to add support for custom types.
Example: serializing IP addresses¶
from ipaddress import IPv4Address, IPv6Address
from typing import Any
from edri.utility.json_encoder import CustomJSONEncoder
class MyJSONEncoder(CustomJSONEncoder):
def default(self, data: Any) -> Any:
if isinstance(data, (IPv4Address, IPv6Address)):
return data.exploded
return super().default(data)
Encoder context¶
Encoders may use self.context to access request-scoped information.
In EDRI, the encoder context is derived from the event itself.
The API layer calls
event.get_context().The result is passed to the JSON encoder as
self.context.
What Event.get_context() returns¶
Event.get_context() returns a dictionary of all keyword-only dataclass fields (kw_only=True) and their current values.
This makes keyword-only fields the recommended place for request-scoped inputs that should be available to JSON encoding (for example values loaded from headers/cookies/scope, or other per-request metadata).
Typical uses:
attaching a database session wrapper
controlling output detail level (public vs internal)
passing authenticated user/session info into model serializers
context is intended to be read by the encoder.
Example: serializing ORM models with a session¶
This pattern:
detects objects that implement
to_json(...)merges them into a SQLAlchemy session
returns a JSON-friendly structure
from typing import Any
from edri.utility.json_encoder import CustomJSONEncoder
from config.database import SessionFactory
from database.models import Base
class SessionJSONEncoder(CustomJSONEncoder):
def default(self, data: Any) -> Any:
if hasattr(data, "to_json") and issubclass(type(data), Base):
# Reuse a request-scoped session passed via context if present,
# otherwise open a short-lived one.
session = (self.context or {}).get("session")
if session is not None:
return session.merge(data, load=True).to_json(session)
with SessionFactory() as session:
return session.merge(data, load=True).to_json(session)
return super().default(data)
A ready-made, base-class-agnostic version ships as an opt-in integration:
from edri.integrations.sqlalchemy import SessionJSONEncoder. Subclass it and implementget_session()to return a context-managedSession, then pass the subclass toedri.start_api(json_encoder=...). It is not a core dependency — install it withpip install edri[sqlalchemy]. See Optional integrations.
Gotchas¶
Always call
super().default(...)for types you do not handle.Keep encoders fast; they run in the response rendering path.
Avoid doing heavy database work per field if you can batch/prepare data earlier.