This is the code I wrote
from typing import Any
from fastapi import APIRouter
from pydantic import BaseModel, ConfigDict, ValidationError
class State(BaseModel):
mode: str | None = None
alarm: int = 0
class StateLoose(State):
model_config = ConfigDict(extra='allow')
class StateExact(State):
model_config = ConfigDict(extra='forbid')
def validated_state(state: dict) -> State:
try:
return StateExact(**state)
except ValidationError as e:
logger.warning("Sanitized input state that caused a validation error. Error: %s", e)
return State(**state)
@router.put("/{client_id}", response_model=State)
def update_state(client_id: str, state: StateLoose) -> Any:
v_state = validated_state(state.dict()).dict()
return update_resource(client_id=client_id, state=v_state)
# Example State inputs
a = {"mode": "MANUAL", "alarm": 1}
b = {"mode": "MANUAL", "alarm": 1, "dog": "bau"}
normal = State(**a)
loose = StateLoose(**a)
exact = StateExact(**a)
From my understanding/tests with/of pydantic
What I wanted to achieve is:
To achieve this I was forced to create 3 different State classes and play with those. Since I plan to have lots of Models, I don't like the idea of having 3 versions of each and it feels like I am doing something quite wrong if it's so weird to accomplish.
Is there a less redundant way to:
You can use a model_validator
with mode="before"
together with the field names from model - i.e. in your validator you see which fields got submitted and log those that you didn't expect.
Setting extra='ignore'
as the model_config
will also make sure that the extra fields gets removed automagically.
from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict, model_validator
class WarnUnknownBase(BaseModel):
model_config = ConfigDict(extra='ignore')
@model_validator(mode='before')
@classmethod
def validate(cls, values):
expected_fields = set(cls.model_fields.keys())
submitted_fields = set(values.keys())
unknown_fields = submitted_fields - expected_fields
if unknown_fields:
print(f"Log these fields: {unknown_fields}")
return values
class Foo(WarnUnknownBase):
foo: int
app = FastAPI()
@app.put('/')
def put_it(foo: Foo):
return foo
Example requests:
λ curl -H "Content-Type: application/json" -X PUT -d "{\"foo\": 42}" http://localhost:8008
{"foo":42}
λ curl -H "Content-Type: application/json" -X PUT -d "{\"foo\": 42, \"bar\": 13}" http://localhost:8008
{"foo":42}
As you can see, bar
gets ignored when the Foo object is created.
The second request will output Log these fields: {'bar'}
on the server side, while the first will keep quiet.
INFO: 127.0.0.1:49458 - "PUT / HTTP/1.1" 200 OK
Log these fields: {'bar'}
INFO: 127.0.0.1:49462 - "PUT / HTTP/1.1" 200 OK