mirror of
https://github.com/almet/notmyidea.git
synced 2025-05-05 23:31:48 +02:00
95 lines
2.6 KiB
Markdown
95 lines
2.6 KiB
Markdown
---
|
|
title: Parsing JSON into Specific Pydantic Models
|
|
tags: pydantic, python, match
|
|
---
|
|
|
|
|
|
I really like [Pydantic](https://docs.pydantic.dev/latest/) because it makes it
|
|
easy to define the the structure of the objects I want to use, using typing.
|
|
|
|
One use case I have at the moment is to parse a json object and build different
|
|
objects depending on some key in the json object.
|
|
|
|
I tried multiple times, and finally managed to do it:
|
|
|
|
have different pydantic classes which share a same property (here named `kind`),
|
|
and end up with the proper classes at the end.
|
|
|
|
Here, I have three types of messages: "OperationMessage", "PeerMessage" and
|
|
"ServerRequest", as follows:
|
|
|
|
```python
|
|
from typing import Literal, Optional
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class OperationMessage(BaseModel):
|
|
"""Message sent from one peer to all the others"""
|
|
|
|
kind: Literal["operation"] = "operation"
|
|
verb: Literal["upsert", "update", "delete"]
|
|
subject: Literal["map", "layer", "feature"]
|
|
metadata: Optional[dict] = None
|
|
key: Optional[str] = None
|
|
|
|
|
|
class PeerMessage(BaseModel):
|
|
"""Message sent from a specific peer to a specific one"""
|
|
|
|
kind: Literal["peermessage"] = "peermessage"
|
|
sender: str
|
|
recipient: str
|
|
# The message can be whatever the peers want. It's not checked by the server.
|
|
message: dict
|
|
|
|
|
|
class ServerRequest(BaseModel):
|
|
"""A request towards the server"""
|
|
|
|
kind: Literal["server"] = "server"
|
|
action: Literal["list-peers"]
|
|
|
|
```
|
|
|
|
Each of these classes share the same `kind` property, which can act as a discriminator.
|
|
|
|
Let's build a generic `Request` class that will be able to build for me the proper objects:
|
|
|
|
```python
|
|
from typing import Union
|
|
from pydantic import Field, RootModel, ValidationError
|
|
|
|
|
|
class Request(RootModel):
|
|
"""Any message coming from the websocket should be one of these, and will be rejected otherwise."""
|
|
|
|
root: Union[ServerRequest, PeerMessage, OperationMessage] = Field(
|
|
discriminator="kind"
|
|
)
|
|
```
|
|
|
|
Which can be used this way:
|
|
|
|
```python
|
|
try:
|
|
incoming = Request.model_validate_json(raw_message)
|
|
except ValidationError as e:
|
|
error = f"An error occurred when receiving the following message: {raw_message}"
|
|
|
|
```
|
|
|
|
And, because we have classes, we can leverage the `match` statement:
|
|
|
|
```python
|
|
match incoming.root:
|
|
case OperationMessage():
|
|
# Here to broadcast the message
|
|
websockets.broadcast(peers, raw_message)
|
|
|
|
case PeerMessage(recipient=_id):
|
|
# Or to send peer messages to the proper peer
|
|
peer = connections.get(_id)
|
|
if peer:
|
|
await peer.send(raw_message)
|
|
|
|
```
|