not_null.py 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344
  1. """Update schemas treat an explicit null for a column that can't be empty as "unchanged"."""
  2. from typing import Any, ClassVar
  3. from pydantic import BaseModel, model_validator
  4. class IgnoresNullForRequiredColumns(BaseModel):
  5. """Base for update (PATCH/PUT) schemas that are applied to one ORM row.
  6. Their fields are mostly ``X | None = None`` so a request can leave them out,
  7. and the API's contract is that null means "unchanged": most routes apply a
  8. field only when it is not None, and the OpenAPI schema allows null. Routes
  9. that apply what was sent with ``model_dump(exclude_unset=True)`` instead
  10. wrote the null into a NOT NULL column and failed with a 500. This drops such
  11. a null before validation, so the field counts as not sent on every route; a
  12. null for a column that may be empty still clears it.
  13. ``__orm_table__`` names the table the schema writes to. It is looked up when
  14. a request is validated, so schemas don't import the models.
  15. """
  16. __orm_table__: ClassVar[str]
  17. @model_validator(mode="before")
  18. @classmethod
  19. def _ignore_null_for_required_columns(cls, data: Any) -> Any:
  20. if not isinstance(data, dict):
  21. return data
  22. columns = _table(cls.__orm_table__).c
  23. by_key = {(info.alias or name): name for name, info in cls.model_fields.items()} | {
  24. name: name for name in cls.model_fields
  25. }
  26. return {
  27. key: value
  28. for key, value in data.items()
  29. if not (value is None and key in by_key and by_key[key] in columns and not columns[by_key[key]].nullable)
  30. }
  31. def _table(name: str):
  32. from backend.app.core.database import load_all_models
  33. return load_all_models().tables[name]