Coverage for fastapi_restly / schemas / _base.py: 94%
540 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-09 18:41 +0000
1import dataclasses
2import functools
3import inspect
4import sys
5import types
6from collections.abc import Iterable, Mapping
7from datetime import datetime
8from typing import (
9 TYPE_CHECKING,
10 Annotated,
11 Any,
12 ClassVar,
13 ForwardRef,
14 Generic,
15 Literal,
16 Optional,
17 Union,
18 cast,
19 final,
20 get_args,
21 get_origin,
22)
24import pydantic
25from pydantic.fields import Field, FieldInfo
26from sqlalchemy import inspect as sa_inspect
27from sqlalchemy import select
28from sqlalchemy.exc import NoResultFound
29from sqlalchemy.ext.asyncio.session import AsyncSession as SA_AsyncSession
30from sqlalchemy.orm.session import Session as SA_Session
31from typing_extensions import TypeAliasType, TypeVar
33from .._mapping import is_mapped_instance
34from .._pagination import (
35 _DEFAULT_PAGINATION,
36 Envelope,
37 NoPagination,
38 NumberedPagination,
39)
40from ..clauses import Unscoped, WhereClause, apply_clauses
41from ..clauses._scopes import _default_scope
42from ..exc import NotFound, RestlyConfigurationError
44if sys.version_info >= (3, 14): 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true
45 import annotationlib
48class BaseSchema(pydantic.BaseModel):
49 """Thin Pydantic base for ORM-facing Restly schemas.
51 Equivalent to::
53 class BaseSchema(pydantic.BaseModel):
54 model_config = pydantic.ConfigDict(from_attributes=True)
56 ``from_attributes=True`` lets Pydantic/FastAPI validate objects by
57 attribute when the schema is used directly. The inherited CRUD endpoint
58 methods still serialize through ``to_single_response()`` so Restly-specific
59 behavior such as ``WriteOnly`` filtering and relationship-id normalization
60 is applied.
61 """
63 model_config = pydantic.ConfigDict(from_attributes=True)
65 @classmethod
66 def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
67 super().__pydantic_init_subclass__(**kwargs)
68 # Reject a ReadOnly/WriteOnly marker buried inside a union member, where
69 # it silently no-ops (see ``_reject_buried_markers``). This fires as the
70 # schema class is defined, so the mistake surfaces at import time. Views
71 # also re-check the schemas they use, to cover schemas that do not derive
72 # from ``BaseSchema``.
73 _reject_buried_markers(cls)
76class _Marker:
77 def __init__(self, name: str):
78 self.name = name
80 def __repr__(self):
81 return f"fr.{self.name}"
84readonly_marker = _Marker("ReadOnly")
85writeonly_marker = _Marker("WriteOnly")
87_T = TypeVar("_T")
89ReadOnly = Annotated[_T, readonly_marker, Field(json_schema_extra={"readOnly": True})]
90# ``exclude=True`` strips the field from serialization at the field level, so it
91# is dropped from every response -- recursively, including in nested schemas, and
92# from the OpenAPI response schema -- while staying a writable request input
93# (exclude does not affect validation). Prefer ``WriteOnly[Optional[T]]`` over
94# ``Optional[WriteOnly[T]]`` / ``WriteOnly[T] | None``: when the marker is only a
95# union member the exclude rides on the inner type and does NOT apply, so the
96# field would leak.
97WriteOnly = Annotated[
98 _T, writeonly_marker, Field(json_schema_extra={"writeOnly": True}, exclude=True)
99]
102# A ReadOnly/WriteOnly marker only takes effect as the OUTER annotation of a
103# field: its ``Annotated`` metadata has to sit at the field's top level. Nested
104# anywhere inside the field's type -- a union member (``Optional[ReadOnly[T]]``,
105# ``WriteOnly[T] | None``) or a container element (``list[WriteOnly[T]]``) -- the
106# marker silently no-ops: ReadOnly fails to drop the field from create/update (it
107# stays writable) and WriteOnly fails to exclude it from responses (it leaks).
108# The guards below detect that misuse and reject it loudly.
109#
110# Coverage boundary: detection reads ``field_info.annotation``, so a marker
111# hidden behind an unresolved forward reference is not seen until the annotation
112# resolves (the view-registration backstop re-checks resolved schemas), and that
113# backstop only inspects a view's own read/create/update schemas -- not a custom
114# ``response_model=`` or a non-``BaseSchema`` nested model. Those narrow cases are
115# left uncovered by design.
116def _annotation_buries_marker(annotation: Any, marker: _Marker) -> bool:
117 """True if ``marker`` appears anywhere below the top level of ``annotation``
118 -- inside a union member, ``Annotated`` inner type, or container arg.
120 Only descends type arguments (``get_args``); it does not recurse into the
121 fields of a nested model, so a nested schema carrying its own top-level
122 marker is not flagged.
123 """
124 if marker in getattr(annotation, "__metadata__", ()):
125 return True
126 return any(_annotation_buries_marker(arg, marker) for arg in get_args(annotation))
129def _find_buried_marker_fields(
130 model_cls: type[pydantic.BaseModel],
131) -> list[tuple[str, _Marker]]:
132 """Return ``(field_name, marker)`` for each field whose ReadOnly/WriteOnly
133 marker is nested inside the field's type instead of wrapping it, where it
134 does not take effect."""
135 buried: list[tuple[str, _Marker]] = []
136 for name, field_info in model_cls.model_fields.items():
137 top_level = getattr(field_info, "metadata", None) or ()
138 for marker in (readonly_marker, writeonly_marker):
139 if marker in top_level:
140 continue
141 if _annotation_buries_marker(field_info.annotation, marker):
142 buried.append((name, marker))
143 return buried
146def _reject_buried_markers(model_cls: type[pydantic.BaseModel]) -> None:
147 """Raise if any field nests a ReadOnly/WriteOnly/RefExists marker inside
148 its type, where the marker machinery cannot see it."""
149 for name, marker in _find_buried_marker_fields(model_cls):
150 raise RestlyConfigurationError(
151 f"{model_cls.__name__}.{name} nests the {marker.name} marker inside "
152 f"its type (e.g. Optional[{marker.name}[T]], {marker.name}[T] | None, "
153 f"or list[{marker.name}[T]]). The marker only takes effect as the "
154 f"outer annotation, so it is silently ignored here and {marker.name} "
155 f"is not applied. Wrap the whole field type instead, e.g. "
156 f"{marker.name}[Optional[T]]."
157 )
158 _reject_buried_ref_exists(model_cls)
161@final
162class _Infer:
163 """Static-only sentinel: the default ``MustExist`` model argument.
165 ``MustExist[int]`` supplies no target model, so the second type parameter
166 defaults to ``_Infer`` (a PEP 696 default that keeps Pyright happy about
167 "supply all params"). It carries no runtime meaning of its own: one type
168 argument is already unambiguous, and the runtime reads it as "resolve the
169 target model from the column's ``ForeignKey``".
170 """
173@final
174class _NotGiven:
175 """Static-only sentinel: ``RefExists`` received no ``scope`` argument.
177 Not given means the target model's ``default_scope`` applies; the
178 explicit opt-out is ``fr.clauses.UNSCOPED``, and ``None`` is rejected
179 so a variable that happens to be ``None`` can never silently unscope.
180 """
183_NOT_GIVEN = _NotGiven()
186class RefExists:
187 """Marker for an existence-checked scalar foreign key.
189 Built by ``MustExist[pk]`` / ``MustExist[pk, Model]`` (or written directly as
190 ``Annotated[pk, RefExists(Model)]``). On write, Restly batch-checks that each
191 marked id exists in the target table and raises ``NotFound`` (404) on a miss.
192 The field stays a plain scalar -- this only adds the check; it does not wrap
193 the value or do any relationship routing.
195 The check applies the target model's ``default_scope``, like
196 ``IDRef``/``IDSchema`` resolution: a reference to a row the scope hides
197 is a miss. ``scope`` overrides that per field: a ``WhereClause`` checks
198 against that clause instead (``RefExists(Item, scope=ItemClauses.trashed)``
199 for a restore target), and ``scope=fr.clauses.UNSCOPED`` checks
200 unscoped, greppably; ``None`` is rejected. A model without a
201 ``default_scope`` is checked unscoped, as before.
203 ``model`` is the target ORM model, or ``_Infer`` when it should be resolved
204 from the marked column's ``ForeignKey`` (the ``MustExist[pk]`` form).
205 """
207 def __init__(
208 self,
209 model: type[Any] | type[_Infer],
210 scope: WhereClause | Unscoped | _NotGiven = _NOT_GIVEN,
211 ) -> None:
212 if scope is None:
213 raise TypeError(
214 "RefExists scope=None says nothing; opt out explicitly with "
215 "scope=fr.clauses.UNSCOPED, so a variable that happens to be "
216 "None can never silently unscope the check"
217 )
218 if not isinstance(scope, (WhereClause, Unscoped, _NotGiven)):
219 raise TypeError(
220 f"RefExists scope must be a WhereClause or UNSCOPED, got "
221 f"{type(scope).__name__}; wrap a raw expression with "
222 "where_clause(), and express a join-dependent predicate "
223 "as EXISTS (.any()/.has())"
224 )
225 self.model = model
226 self.scope = scope
228 def __repr__(self) -> str:
229 name = (
230 "infer"
231 if self.model is _Infer
232 else getattr(self.model, "__name__", self.model)
233 )
234 if isinstance(self.scope, _NotGiven): 234 ↛ 236line 234 didn't jump to line 236 because the condition on line 234 was always true
235 return f"RefExists({name})"
236 return f"RefExists({name}, scope={self.scope!r})"
239def _annotation_contains_ref_exists(annotation: Any) -> bool:
240 if any(isinstance(m, RefExists) for m in getattr(annotation, "__metadata__", ())):
241 return True
242 return any(_annotation_contains_ref_exists(arg) for arg in get_args(annotation))
245def _reject_buried_ref_exists(model_cls: type[pydantic.BaseModel]) -> None:
246 """Raise for a RefExists marker the check machinery cannot see.
248 The existence check reads a field's top-level metadata (plus the
249 Optional-unwrapped form); a marker inside a container element
250 (``list[MustExist[int, T]]``) is invisible there, so the field would
251 validate and write with no check at all -- a silent no-op of a check
252 the schema explicitly asks for.
253 """
254 for name, field_info in model_cls.model_fields.items():
255 if not _annotation_contains_ref_exists(field_info.annotation):
256 continue
257 if _ref_exists_marker(field_info) is not None:
258 continue
259 raise RestlyConfigurationError(
260 f"{model_cls.__name__}.{name} buries a RefExists/MustExist marker "
261 f"inside its type (e.g. list[MustExist[int, T]]), where the "
262 f"existence check cannot see it, so the reference would go "
263 f"unchecked. To reference several rows, use "
264 f"list[IDRef[T]] on a relationship-named field instead."
265 )
268class TimestampsSchemaMixin(pydantic.BaseModel):
269 created_at: ReadOnly[datetime]
270 updated_at: ReadOnly[datetime]
273# Unbound: a mapped class need not subclass DeclarativeBase.
274SQLAlchemyModel = TypeVar("SQLAlchemyModel", default=Any)
275_IDREF_UNSET = object()
276_SCHEMA_RESOURCE_SUFFIXES = ("Schema", "Response")
279@functools.cache
280def _id_type_adapter(id_type: Any) -> pydantic.TypeAdapter[Any]:
281 return pydantic.TypeAdapter(id_type)
284def _schema_resource_name(model_cls: type[pydantic.BaseModel]) -> str:
285 """The resource part of a schema's class name: the name without a final
286 ``Schema`` or ``Response``. ``UserSchema`` and ``UserResponse`` both give
287 ``User``."""
288 name = model_cls.__name__
289 for suffix in _SCHEMA_RESOURCE_SUFFIXES:
290 if name.endswith(suffix) and len(name) > len(suffix):
291 return name[: -len(suffix)]
292 return name
295def _schema_role_name(model_cls: type[pydantic.BaseModel], role: str) -> str:
296 """The name of a class that Restly generates from a schema:
297 ``<Resource><Role>``, as in ``UserCreate``."""
298 return f"{_schema_resource_name(model_cls)}{role}"
301def _own_annotations(cls: type) -> dict[str, Any]:
302 """Return the annotations ``cls`` itself declares, without failing on them.
304 Under PEP 563 the values are strings. On 3.14 a name that is undefined at
305 runtime, such as a ``TYPE_CHECKING`` import, comes back as a ``ForwardRef``
306 where reading ``__annotations__`` would raise ``NameError``.
307 """
308 if sys.version_info >= (3, 14): 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 return annotationlib.get_annotations(
310 cls, format=annotationlib.Format.FORWARDREF
311 )
312 return inspect.get_annotations(cls)
315def _is_unresolved(annotation: Any) -> bool:
316 """Whether ``annotation`` still holds a string or ``ForwardRef`` anywhere."""
317 if isinstance(annotation, (str, ForwardRef)):
318 return True
319 origin = get_origin(annotation)
320 if origin is Literal: 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 return False
322 args = get_args(annotation)
323 if origin is Annotated: 323 ↛ 325line 323 didn't jump to line 325 because the condition on line 323 was never true
324 # Metadata may be any object, strings included.
325 args = args[:1]
326 return any(_is_unresolved(arg) for arg in args)
329def _model_id_type(sql_model: type[Any]) -> Any:
330 """Return the Python type of ``sql_model``'s ``id`` primary key, or ``None``.
332 Reads the ``id`` annotation off the model's MRO (works before the mapper is
333 configured), then falls back to the SA mapper -- which also covers a PEP 563
334 / ``from __future__ import annotations`` model, whose ``id`` annotation is a
335 string the mapper resolves authoritatively. Shared by ``IDSchema``/``IDRef``
336 (via ``_get_sql_model_id_type``) and ``MustExist``.
337 """
338 for model_cls in sql_model.mro():
339 annotation = _own_annotations(model_cls).get("id")
340 if annotation is None:
341 continue
342 if get_origin(annotation) is not None:
343 args = get_args(annotation)
344 if args: 344 ↛ 346line 344 didn't jump to line 346 because the condition on line 344 was always true
345 annotation = args[0]
346 if _is_unresolved(annotation):
347 # Don't return a raw string; the mapper below resolves the real
348 # column type without eval'ing it.
349 break
350 return annotation
352 # Fallback: ask the SA mapper. `python_type` raises NotImplementedError for
353 # column types without a Python equivalent (e.g. some user types), and
354 # a class that is not mapped has no mapper to ask.
355 try:
356 return sa_inspect(sql_model, raiseerr=False).primary_key[0].type.python_type
357 except (AttributeError, NotImplementedError, IndexError):
358 return None
361class IDSchema(BaseSchema, Generic[SQLAlchemyModel]):
362 """Response-schema base that adds a read-only ``id``; parametrized as a
363 field type, a nested-object relationship reference.
365 - As a BASE CLASS (``class UserSchema(IDSchema): ...``) it adds the resource's
366 own read-only ``id`` -- the common use.
367 - As a FIELD TYPE on a RELATIONSHIP-named field (``author: IDSchema[User]``)
368 the wire format is ``{"id": N}`` (JSON-API / React-Admin); Restly resolves
369 the id to the related ORM object, so ``data.author`` is a wrapper (read
370 ``.id``). Use ``IDRef[User]`` for flat-id wire (``5``) instead.
372 For a scalar foreign-key COLUMN (``author_id``), use ``fr.MustExist[int, T]``
373 rather than ``IDSchema``/``IDRef``: it keeps the field a plain checked id
374 instead of turning a ``*_id`` field into an object wrapper.
375 """
377 # Keep this broad so relation-id payloads can target non-int primary keys.
378 id: ReadOnly[Any]
380 @classmethod
381 def _get_sql_model_annotation(cls) -> type[Any] | None:
382 # `__pydantic_generic_metadata__` is set on parameterised subclasses;
383 # on the bare `IDSchema` class the "args" tuple may be missing or empty.
384 try:
385 sql_model = cls.__pydantic_generic_metadata__["args"][0]
386 except (KeyError, IndexError, TypeError):
387 return None
388 return sql_model if isinstance(sql_model, type) else None
390 @classmethod
391 def _get_sql_model_id_type(cls) -> Any:
392 sql_model = cls._get_sql_model_annotation()
393 if sql_model is None:
394 return None
395 return _model_id_type(sql_model)
397 @pydantic.field_validator("id", mode="before", check_fields=False)
398 @classmethod
399 def _coerce_id_to_model_primary_key_type(cls, value: Any) -> Any:
400 id_type = cls._get_sql_model_id_type()
401 if id_type in (None, Any):
402 return value
403 return _id_type_adapter(id_type).validate_python(value)
405 @pydantic.model_validator(mode="before")
406 @classmethod
407 def _coerce_scalar(cls, v: Any) -> Any:
408 # A pure id reference (``IDSchema[Model]`` / ``IDRef[Model]``, whose only
409 # field is ``id``) accepts a bare scalar id or a related ORM row where the
410 # ``{"id": ...}`` mapping is expected, so the reference type is
411 # self-sufficient under plain ``from_attributes``: a scalar FK column read
412 # straight off the row in ``to_single_response``, a related row reached
413 # through a relationship, or a ``{"id": N}`` payload all validate without
414 # any view-layer pre-extraction. A subclass that adds fields (a nested
415 # response schema) is NOT a pure reference, so it validates normally and
416 # its row is never collapsed to just the id.
417 if set(cls.model_fields) != {"id"}:
418 return v
419 if isinstance(v, dict):
420 return v
421 if is_mapped_instance(v):
422 return {"id": getattr(v, "id")}
423 return {"id": v}
425 def get_sql_model_annotation(self) -> type[SQLAlchemyModel] | None:
426 """
427 Return the annotation on IDSchema when used as:
429 foo: IDSchema[Foo]
431 This property will return "Foo".
432 """
433 # The runtime introspection returns the bound type; cast through the
434 # generic parameter so callers see the concrete model class.
435 return self._get_sql_model_annotation() # type: ignore[return-value]
438class IDRef(IDSchema[SQLAlchemyModel], Generic[SQLAlchemyModel]):
439 """Flat-id reference to a related row, for a RELATIONSHIP-named field.
441 Use ``IDRef[T]`` when the field is named after a relationship
442 (``author: IDRef[User]``, ``products: list[IDRef[Product]]``), not after a
443 ``*_id`` column. The wire format is the raw id (``5``); input also accepts
444 ``{"id": N}``. Restly resolves the id to the related ORM object, so
445 ``data.author`` is an ``IDRef`` wrapper (read ``.id``), not a plain scalar.
447 For a scalar foreign-key COLUMN (``author_id``, ``task_id``) use
448 ``fr.MustExist[int, T]`` instead -- it keeps the field a plain checked id
449 (``data.author_id == 1``), rather than making a ``*_id`` field an object
450 wrapper. For nested ``{"id": N}`` wire on a relationship, use ``IDSchema[T]``.
452 author: IDRef[User] # to-one relationship, flat id
453 products: list[IDRef[Product]] # serializes as [1, 2, 3]
455 Resolution applies the referenced model's ``default_scope``: a
456 reference to a row the scope hides raises ``NotFound``, so a
457 tenant scope declared once covers every reference to the model. A model
458 without a ``default_scope`` resolves by bare primary key; gate visibility
459 there in ``authorize`` (``data.<field>.id`` is the requested id, before
460 resolution) or ``before_action_commit`` (the resolved row is on the built
461 object). See the Foreign Keys and Relationships how-to, "Visibility and
462 Multi-Tenancy".
463 """
465 def __init__(self, value: Any = _IDREF_UNSET, **data: Any) -> None:
466 if value is not _IDREF_UNSET:
467 if data: 467 ↛ 468line 467 didn't jump to line 468 because the condition on line 467 was never true
468 raise TypeError(
469 "IDRef accepts either a positional id or keyword fields"
470 )
471 data = value if isinstance(value, dict) else {"id": value}
472 super().__init__(**data)
474 @classmethod
475 def __get_pydantic_json_schema__(
476 cls, core_schema: Any, handler: pydantic.GetJsonSchemaHandler
477 ) -> dict[str, Any]:
478 id_type = cls._get_sql_model_id_type()
479 if id_type in (None, Any): 479 ↛ 480line 479 didn't jump to line 480 because the condition on line 479 was never true
480 return {}
481 return pydantic.TypeAdapter(id_type).json_schema(
482 mode=getattr(handler, "mode", "validation")
483 )
485 @pydantic.model_serializer
486 def _serialize_flat(self) -> Any:
487 return self.id if hasattr(self, "id") else self
490if TYPE_CHECKING:
491 _MustExistPK = TypeVar("_MustExistPK")
492 _MustExistModel = TypeVar("_MustExistModel", default=_Infer)
494 # Static view: ``MustExist[pk]`` and ``MustExist[pk, Model]`` both read as the
495 # pk scalar (the alias value is the first param), so a field stays a plain
496 # scalar -- ``data.post_id`` is an ``int``, not a wrapper. The target model is
497 # a runtime-only concern (second param, defaulting to ``_Infer``); the runtime
498 # ``MustExist`` is defined below.
499 MustExist = TypeAliasType(
500 "MustExist", _MustExistPK, type_params=(_MustExistPK, _MustExistModel)
501 )
502else:
504 class MustExist:
505 """Existence-checked scalar foreign key.
507 ``MustExist[int]`` is a checked ``int`` foreign key; the target model is
508 inferred from the column's ``ForeignKey``. Spell the model out with a
509 second argument when you want it explicit (``MustExist[int, Post]``), and
510 use the first for a non-int primary key (``MustExist[UUID, Account]``).
512 Unlike ``IDRef``/``IDSchema`` this is **not** a wrapper: the field stays
513 the pk scalar everywhere (wire, column, ``data.<field>``), plus a batched
514 existence check on write (404 on a miss). The check applies the target
515 model's ``default_scope``; check against another clause, or none, by
516 writing the marker form ``Annotated[<pk>, RefExists(Model, scope=...)]``
517 directly (see :class:`RefExists`).
518 """
520 def __class_getitem__(cls, params: Any) -> Any:
521 pk_type, model = params if isinstance(params, tuple) else (params, _Infer)
522 return Annotated[pk_type, RefExists(model)]
525def _unwrap_optional_annotation(annotation: Any) -> Any:
526 """Unwrap Optional[X] or X | None to X. Returns annotation unchanged otherwise."""
527 origin = get_origin(annotation)
528 if origin not in (types.UnionType, Union):
529 return annotation
531 non_none_args = [arg for arg in get_args(annotation) if arg is not type(None)]
532 if len(non_none_args) == 1: 532 ↛ 534line 532 didn't jump to line 534 because the condition on line 532 was always true
533 return non_none_args[0]
534 return annotation
537def is_reference_annotation(annotation: Any) -> bool:
538 """True if ``annotation`` is an ``IDRef``/``IDSchema`` reference type.
540 The canonical reference-type check, kept beside the types so callers (views,
541 query) don't re-introspect Pydantic generics themselves. Unwraps ``Optional``
542 first, then matches the bare types or a parametrization whose generic origin
543 is ``IDSchema``/``IDRef`` (``IDRef[Post]``, ``IDSchema[Post]``) -- not a user
544 subclass that merely inherits from them.
545 """
546 annotation = _unwrap_optional_annotation(annotation)
547 if annotation in (IDSchema, IDRef):
548 return True
549 if not inspect.isclass(annotation):
550 return False
551 try:
552 if not issubclass(annotation, IDSchema):
553 return False
554 except TypeError:
555 return False
556 metadata = getattr(annotation, "__pydantic_generic_metadata__", {})
557 return metadata.get("origin") in (IDSchema, IDRef)
560def is_reference_field(schema_cls: type[pydantic.BaseModel], field_name: str) -> bool:
561 """True if ``schema_cls``'s ``field_name`` is typed as an ``IDRef``/``IDSchema``."""
562 field_info = schema_cls.model_fields.get(field_name)
563 if field_info is None: 563 ↛ 564line 563 didn't jump to line 564 because the condition on line 563 was never true
564 return False
565 return is_reference_annotation(field_info.annotation)
568def reference_origin_and_target(annotation: Any) -> tuple[type, type | None] | None:
569 """For an ``IDRef``/``IDSchema`` field annotation (``Optional`` unwrapped),
570 return ``(origin, target)``: the ``IDRef``/``IDSchema`` class and the
571 referenced model (``None`` for a bare, unparametrized reference), or ``None``
572 when the annotation is not a reference type. Lets the scalar-named-reference
573 lint name both the type and its target in the ``MustExist`` hint.
574 """
575 annotation = _unwrap_optional_annotation(annotation)
576 if annotation in (IDSchema, IDRef): 576 ↛ 577line 576 didn't jump to line 577 because the condition on line 576 was never true
577 return (annotation, None)
578 if not inspect.isclass(annotation): 578 ↛ 579line 578 didn't jump to line 579 because the condition on line 578 was never true
579 return None
580 try:
581 if not issubclass(annotation, IDSchema):
582 return None
583 except TypeError:
584 return None
585 metadata = getattr(annotation, "__pydantic_generic_metadata__", {})
586 origin = metadata.get("origin")
587 if origin not in (IDSchema, IDRef): 587 ↛ 588line 587 didn't jump to line 588 because the condition on line 587 was never true
588 return None
589 args = metadata.get("args") or ()
590 target = args[0] if args and isinstance(args[0], type) else None
591 return (origin, target)
594async def _async_resolve_ids_to_sqlalchemy_objects(
595 session: SA_AsyncSession, schema_obj: pydantic.BaseModel
596) -> dict[str, Any]:
597 """
598 Resolve any IDSchema reference fields on ``schema_obj`` to SQLAlchemy rows.
599 A database request is made for each IDSchema to look up the related row in the database.
600 If an id is not found in the database `sqlalchemy.orm.exc.NoResultFound` is raised.
602 Returns a ``{field_name: resolved_object_or_list}`` mapping for the fields
603 that referenced a model; ``schema_obj`` itself is left unmodified, so it
604 keeps its validated wire shape (``IDRef[T]`` values, not ORM rows). The
605 write path consumes the returned mapping.
607 The lookup applies the referenced model's ``default_scope``, so a
608 reference to a row the scope hides is ``NotFound``. A
609 model without a ``default_scope`` resolves by bare primary key, as before;
610 see the ``IDRef`` docstring.
611 """
612 # Go over all Pydantic fields and check if any of them are an IDSchema object or
613 # a list of IDSchema objects.
614 resolved: dict[str, Any] = {}
615 for field in schema_obj.model_fields_set:
616 value = getattr(schema_obj, field, None)
618 if isinstance(value, IDSchema):
619 sql_model = value.get_sql_model_annotation()
620 if not sql_model:
621 continue
623 scope = _default_scope(sql_model)
624 if scope is None:
625 # unscoped: keep get_one, which serves identity-map hits
626 try:
627 sql_model_obj = await session.get_one(sql_model, value.id)
628 except NoResultFound as e:
629 raise NotFound(f"Id not found for {field}: {value.id}") from e
630 else:
631 # mapper pk, not `.id`: the pk attribute can be named anything
632 pk_col = sa_inspect(sql_model).primary_key[0]
633 query = apply_clauses(
634 select(sql_model).where(pk_col == value.id), scope
635 )
636 sql_model_obj = (await session.scalars(query)).first()
637 if sql_model_obj is None:
638 raise NotFound(f"Id not found for {field}: {value.id}")
639 resolved[field] = sql_model_obj
641 elif isinstance(value, list) and any(isinstance(i, IDSchema) for i in value):
642 # Assume all IdSchemas are for the same model
643 sql_model = value[0].get_sql_model_annotation()
644 if not sql_model: 644 ↛ 645line 644 didn't jump to line 645 because the condition on line 644 was never true
645 continue
647 # Resolve via an id -> row map: keeps the client's order (first
648 # appearance, deduped) and makes a missing id one absent from the map,
649 # so a repeated id can't spuriously 404 (``IN`` dedups and reorders).
650 ids = [obj.id for obj in value]
651 unique_ids = list(dict.fromkeys(ids))
652 query = select(sql_model).where(sql_model.id.in_(unique_ids))
653 scope = _default_scope(sql_model)
654 if scope is not None: 654 ↛ 655line 654 didn't jump to line 655 because the condition on line 654 was never true
655 query = apply_clauses(query, scope)
656 by_id = {o.id: o for o in await session.scalars(query)}
658 missing = [i for i in unique_ids if i not in by_id]
659 if missing:
660 raise NotFound(f"Id not found for {field}: {missing}")
662 resolved[field] = [by_id[i] for i in unique_ids]
664 return resolved
667def _resolve_ids_to_sqlalchemy_objects(
668 session: SA_Session, schema_obj: pydantic.BaseModel
669) -> dict[str, Any]:
670 """
671 Resolve any IDSchema reference fields on ``schema_obj`` to SQLAlchemy rows.
672 A database request is made for each IDSchema to look up the related row in the database.
673 If an id is not found in the database `sqlalchemy.orm.exc.NoResultFound` is raised.
675 Returns a ``{field_name: resolved_object_or_list}`` mapping for the fields
676 that referenced a model; ``schema_obj`` itself is left unmodified, so it
677 keeps its validated wire shape (``IDRef[T]`` values, not ORM rows). The
678 write path consumes the returned mapping.
680 The lookup applies the referenced model's ``default_scope``, so a
681 reference to a row the scope hides is ``NotFound``. A
682 model without a ``default_scope`` resolves by bare primary key, as before;
683 see the ``IDRef`` docstring.
684 """
685 # Go over all Pydantic fields and check if any of them are an IDSchema object or
686 # a list of IDSchema objects.
687 resolved: dict[str, Any] = {}
688 for field in schema_obj.model_fields_set:
689 value = getattr(schema_obj, field, None)
691 if isinstance(value, IDSchema):
692 sql_model = value.get_sql_model_annotation()
693 if not sql_model:
694 continue
696 scope = _default_scope(sql_model)
697 if scope is None:
698 # unscoped: keep get_one, which serves identity-map hits
699 try:
700 sql_model_obj = session.get_one(sql_model, value.id)
701 except NoResultFound as e:
702 raise NotFound(f"Id not found for {field}: {value.id}") from e
703 else:
704 # mapper pk, not `.id`: the pk attribute can be named anything
705 pk_col = sa_inspect(sql_model).primary_key[0]
706 query = apply_clauses(
707 select(sql_model).where(pk_col == value.id), scope
708 )
709 sql_model_obj = session.scalars(query).first()
710 if sql_model_obj is None:
711 raise NotFound(f"Id not found for {field}: {value.id}")
712 resolved[field] = sql_model_obj
714 elif isinstance(value, list) and any(isinstance(i, IDSchema) for i in value):
715 # Assume all IdSchemas are for the same model
716 sql_model = value[0].get_sql_model_annotation()
717 if not sql_model: 717 ↛ 718line 717 didn't jump to line 718 because the condition on line 717 was never true
718 continue
720 # Resolve via an id -> row map: keeps the client's order (first
721 # appearance, deduped) and makes a missing id one absent from the map,
722 # so a repeated id can't spuriously 404 (``IN`` dedups and reorders).
723 ids = [obj.id for obj in value]
724 unique_ids = list(dict.fromkeys(ids))
725 query = select(sql_model).where(sql_model.id.in_(unique_ids))
726 scope = _default_scope(sql_model)
727 if scope is not None:
728 query = apply_clauses(query, scope)
729 by_id = {o.id: o for o in session.scalars(query)}
731 missing = [i for i in unique_ids if i not in by_id]
732 if missing:
733 raise NotFound(f"Id not found for {field}: {missing}")
735 resolved[field] = [by_id[i] for i in unique_ids]
737 return resolved
740def _ref_exists_marker(field_info: FieldInfo) -> RefExists | None:
741 """Return the ``RefExists`` marker on a field, or ``None``.
743 Found at the field's top level (``MustExist[pk, M]`` / ``Annotated[pk,
744 RefExists(M)]``), or recovered from the union member for an optional field
745 (``MustExist[pk, M] | None``), where Pydantic keeps the marker on the inner
746 annotation rather than ``field_info.metadata`` -- so an optional checked FK
747 is checked like the plain form (parity with ``Optional[IDRef]``).
748 """
749 for m in field_info.metadata:
750 if isinstance(m, RefExists):
751 return m
752 unwrapped = _unwrap_optional_annotation(field_info.annotation)
753 for m in getattr(unwrapped, "__metadata__", ()):
754 if isinstance(m, RefExists): 754 ↛ 753line 754 didn't jump to line 753 because the condition on line 754 was always true
755 return m
756 return None
759def _infer_ref_model(model_cls: type[Any], field_name: str) -> type[Any]:
760 """Resolve the target model for a ``MustExist[pk]`` field from its FK column.
762 ``MustExist[pk]`` (one argument) leaves the model to be inferred: the column
763 named by the field must carry exactly one ``ForeignKey``, whose table maps to
764 the target model. Raises ``RestlyConfigurationError`` when the field does not
765 map to a single-FK column -- name the model explicitly (``MustExist[pk,
766 Model]``) there.
767 """
768 model_mapper = sa_inspect(model_cls)
769 column = model_mapper.columns.get(field_name)
770 foreign_keys = list(column.foreign_keys) if column is not None else []
771 if len(foreign_keys) == 1:
772 target_table = foreign_keys[0].column.table
773 for mapper in model_mapper.registry.mappers: 773 ↛ 776line 773 didn't jump to line 776 because the loop on line 773 didn't complete
774 if mapper.local_table is target_table:
775 return mapper.class_
776 raise RestlyConfigurationError(
777 f"{model_cls.__name__}.{field_name}: MustExist[...] with one argument "
778 f"infers the target model from the column's ForeignKey, but its FK "
779 f"target table '{target_table.name}' has no mapped model. "
780 f"Name it explicitly: MustExist[<pk>, <Model>]."
781 )
782 raise RestlyConfigurationError(
783 f"{model_cls.__name__}.{field_name}: MustExist[...] with one argument "
784 f"infers the target model from the column's ForeignKey, but "
785 f"'{field_name}' does not map to a column with exactly one foreign key. "
786 f"Name it explicitly: MustExist[<pk>, <Model>]."
787 )
790def _effective_ref_scope(marker: RefExists, model: type[Any]) -> WhereClause | None:
791 """The scope a ``RefExists`` check applies: the marker's own ``scope``
792 when given (``UNSCOPED`` meaning explicitly none), the target model's
793 ``default_scope`` otherwise."""
794 if isinstance(marker.scope, _NotGiven):
795 return _default_scope(model)
796 if isinstance(marker.scope, Unscoped):
797 return None
798 return marker.scope
801def _ref_exists_fields(
802 model_cls: type[Any], schema_obj: pydantic.BaseModel
803) -> dict[tuple[type[Any], WhereClause | None], list[tuple[str, Any]]]:
804 """Group provided, non-``None`` ``RefExists``-marked scalar fields by
805 target model and effective scope.
807 Returns ``{(model, scope): [(field_name, id), ...]}`` for fields whose
808 annotation carries a ``RefExists`` marker (``MustExist[pk]`` /
809 ``MustExist[pk, Model]``, or an explicit ``Annotated[pk, RefExists(Model)]``).
810 A marker left to infer (``MustExist[pk]``) is resolved to the model behind
811 the column's ``ForeignKey`` on ``model_cls``; the scope is resolved per
812 field (see :func:`_effective_ref_scope`), so two fields on one model can
813 check against different scopes. Only fields actually supplied
814 (``model_fields_set``) with a non-``None`` value are included.
815 """
816 by_target: dict[tuple[type[Any], WhereClause | None], list[tuple[str, Any]]] = {}
817 model_fields = type(schema_obj).model_fields
818 for field_name in schema_obj.model_fields_set:
819 field_info = model_fields.get(field_name)
820 if field_info is None: 820 ↛ 821line 820 didn't jump to line 821 because the condition on line 820 was never true
821 continue
822 marker = _ref_exists_marker(field_info)
823 if marker is None:
824 continue
825 value = getattr(schema_obj, field_name, None)
826 if value is None:
827 continue
828 model = marker.model
829 if model is _Infer:
830 model = _infer_ref_model(model_cls, field_name)
831 scope = _effective_ref_scope(marker, model)
832 by_target.setdefault((model, scope), []).append((field_name, value))
833 return by_target
836def _raise_for_missing_refs(items: list[tuple[str, Any]], found: set[Any]) -> None:
837 for field_name, value in items:
838 if value not in found:
839 raise NotFound(f"Id not found for {field_name}: {value}")
842def _ref_exists_query(
843 model: type[Any], scope: WhereClause | None, items: list[tuple[str, Any]]
844) -> Any:
845 unique = list(dict.fromkeys(value for _, value in items))
846 mapper = sa_inspect(model)
847 # the mapped attribute, not the Core column: an ORM statement, so a
848 # session-level rule (with_loader_criteria) reaches this check too
849 pk = getattr(model, mapper.get_property_by_column(mapper.primary_key[0]).key)
850 query = select(pk).where(pk.in_(unique))
851 if scope is not None:
852 query = apply_clauses(query, scope)
853 return query
856def _check_ref_exists(
857 session: SA_Session, model_cls: type[Any], schema_obj: pydantic.BaseModel
858) -> None:
859 """Batch-validate that every ``RefExists``-marked id exists (sync).
861 One ``SELECT pk WHERE pk IN (...)`` per (referenced model, scope) pair
862 (no N+1); raises ``NotFound`` (404) naming the field and id on a miss.
863 The scope is the marker's own when given, the target model's
864 ``default_scope`` otherwise, so a row the scope hides is a miss.
865 ``model_cls`` is the model being written, used to infer the target of a
866 ``MustExist[pk]`` field.
867 """
868 for (model, scope), items in _ref_exists_fields(model_cls, schema_obj).items():
869 found = set(session.scalars(_ref_exists_query(model, scope, items)))
870 _raise_for_missing_refs(items, found)
873async def _async_check_ref_exists(
874 session: SA_AsyncSession, model_cls: type[Any], schema_obj: pydantic.BaseModel
875) -> None:
876 """Async twin of :func:`_check_ref_exists`."""
877 for (model, scope), items in _ref_exists_fields(model_cls, schema_obj).items():
878 found = set(await session.scalars(_ref_exists_query(model, scope, items)))
879 _raise_for_missing_refs(items, found)
882def get_read_only_fields(model_cls: type[pydantic.BaseModel]) -> list[str]:
883 """Get all fields from a model annotated as ReadOnly[]"""
884 read_only_fields: list[str] = []
885 # Get read-only fields from Annotated metadata
886 for field_name, field_info in model_cls.model_fields.items():
887 metadata = getattr(field_info, "metadata", None)
888 if metadata and readonly_marker in metadata:
889 read_only_fields.append(field_name)
890 return read_only_fields
893def is_readonly_field(
894 model: pydantic.BaseModel | type[pydantic.BaseModel], field_name: str
895) -> bool:
896 """Check if a specific field is marked as readonly."""
897 if isinstance(model, pydantic.BaseModel):
898 model = model.__class__
899 field_info = model.model_fields.get(field_name)
900 return _is_readonly(field_info)
903def _is_readonly(field_info: FieldInfo | None) -> bool:
904 if field_info is None:
905 return False
906 metadata = getattr(field_info, "metadata", None)
907 if not metadata:
908 return False
909 return readonly_marker in metadata
912def _is_writeonly(field_info: FieldInfo | None) -> bool:
913 if field_info is None:
914 return False
915 metadata = getattr(field_info, "metadata", None)
916 if not metadata:
917 return False
918 return writeonly_marker in metadata
921def get_write_only_fields(model_cls: type[pydantic.BaseModel]) -> list[str]:
922 """Get all fields from a model annotated as WriteOnly[]"""
923 write_only_fields: list[str] = []
924 # Get write-only fields from Annotated metadata
925 for field_name, field_info in model_cls.model_fields.items():
926 if _is_writeonly(field_info):
927 write_only_fields.append(field_name)
928 return write_only_fields
931def is_writeonly_field(
932 model_cls: pydantic.BaseModel | type[pydantic.BaseModel], field_name: str
933) -> bool:
934 """Check if a specific field is marked as writeonly."""
935 if isinstance(model_cls, pydantic.BaseModel):
936 model_cls = model_cls.__class__
937 field_info = model_cls.model_fields.get(field_name)
938 return _is_writeonly(field_info)
941def create_model_without_read_only_fields(
942 model_cls: type[pydantic.BaseModel], *, omit: Iterable[str] = ()
943) -> type[pydantic.BaseModel]:
944 """
945 Create a subclass of the given pydantic model class with a new name.
946 Read-only fields are removed, and so are the fields named in ``omit``.
947 """
948 new_model_name = _schema_role_name(model_cls, "Create")
949 new_doc = (model_cls.__doc__ or "") + "\nRead-only fields have been removed."
951 # Create a subclass that mixes in OmitReadOnlyMixin
952 new_model_cls = type(
953 new_model_name,
954 (OmitReadOnlyMixin, model_cls),
955 {
956 "__module__": model_cls.__module__,
957 "__doc__": new_doc,
958 "__restly_omit_fields__": frozenset(omit),
959 },
960 )
962 return new_model_cls
965class OmitReadOnlyMixin(pydantic.BaseModel):
966 """
967 Mixin for pydantic models that removes all fields marked as ReadOnly.
969 Implementation note: this mutates ``cls.model_fields`` in place and then
970 calls ``model_rebuild(force=True)`` to regenerate the validator/serializer.
971 Pydantic v2 does not officially document mutation of ``model_fields`` as
972 a supported customisation hook, but this approach has been stable since
973 pydantic 2.0 and works on the pinned minimum (``pydantic>=2.11.7``). If
974 a future pydantic release freezes the dict, switch to constructing a new
975 model via ``pydantic.create_model(...)`` over the kept fields. The
976 regression test ``tests/test_pydantic_model_fields_mutation.py`` exercises
977 this contract on the currently-installed pydantic.
978 """
980 @classmethod
981 def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
982 super().__pydantic_init_subclass__(**kwargs)
984 # Collect readonly fields to delete first
985 omit = getattr(cls, "__restly_omit_fields__", ())
986 readonly_fields = []
987 for name, field_info in cls.model_fields.items():
988 if _is_readonly(field_info) or name in omit:
989 readonly_fields.append(name)
991 # Delete readonly fields after iteration is complete
992 for name in readonly_fields:
993 del cls.model_fields[name]
995 cls.model_rebuild(force=True)
998class _OmitWriteOnlyMixin(pydantic.BaseModel):
999 """Mixin for the response class: removes the WriteOnly fields."""
1001 #: The names and aliases of the removed WriteOnly fields.
1002 __restly_writeonly_keys__: ClassVar[frozenset[str]] = frozenset()
1004 @classmethod
1005 def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
1006 super().__pydantic_init_subclass__(**kwargs)
1008 writeonly_fields = {
1009 name: field_info
1010 for name, field_info in cls.model_fields.items()
1011 if _is_writeonly(field_info)
1012 }
1013 if not writeonly_fields:
1014 return
1015 cls.__restly_writeonly_keys__ = frozenset(
1016 key
1017 for name, field_info in writeonly_fields.items()
1018 for key in _input_keys(name, field_info)
1019 )
1020 for name in writeonly_fields:
1021 del cls.model_fields[name]
1022 _skip_field_checks(cls, set(writeonly_fields))
1024 # A forward reference that is not defined yet does not fail here:
1025 # pydantic builds the class again when it is first used.
1026 cls.model_rebuild(force=True, raise_errors=False)
1028 # "before", not "wrap": pydantic does not pass ``by_name`` through a wrap
1029 # validator to the nested models.
1030 @pydantic.model_validator(mode="before")
1031 @classmethod
1032 def _drop_writeonly_keys(cls, value: Any) -> Any:
1033 """Drop WriteOnly keys from a dict. With ``extra="allow"`` they would
1034 otherwise go out as extra fields."""
1035 keys = cls.__restly_writeonly_keys__
1036 if keys and isinstance(value, Mapping):
1037 return {key: item for key, item in value.items() if key not in keys}
1038 return value
1041def _input_keys(name: str, field_info: FieldInfo) -> set[str]:
1042 """The keys under which a field can be in a dict: its name and aliases."""
1043 keys = {name}
1044 if field_info.alias is not None:
1045 keys.add(field_info.alias)
1046 validation_alias = field_info.validation_alias
1047 if isinstance(validation_alias, str):
1048 keys.add(validation_alias)
1049 elif isinstance(validation_alias, pydantic.AliasChoices): 1049 ↛ 1050line 1049 didn't jump to line 1050 because the condition on line 1049 was never true
1050 keys.update(
1051 choice for choice in validation_alias.choices if isinstance(choice, str)
1052 )
1053 return keys
1056def _skip_field_checks(cls: type[pydantic.BaseModel], removed: set[str]) -> None:
1057 """Let validators and serializers of the removed fields stay on ``cls``.
1059 Pydantic checks that each field validator and serializer names a field of
1060 the class. A ``@field_validator("password")`` on the view's schema would
1061 then fail the response class, which has no ``password``. Without the
1062 check, pydantic does not use it there.
1063 """
1064 decorators = cls.__pydantic_decorators__
1065 for group in (
1066 decorators.validators,
1067 decorators.field_validators,
1068 decorators.field_serializers,
1069 ):
1070 for name, decorator in list(group.items()):
1071 if removed.intersection(decorator.info.fields):
1072 info = dataclasses.replace(decorator.info, check_fields=False)
1073 group[name] = dataclasses.replace(decorator, info=info) # type: ignore[assignment]
1076def _as_response(cls: type[pydantic.BaseModel], value: Any) -> Any:
1077 """An instance of another model as an instance of the response class
1078 ``cls``. Fields that ``cls`` does not have stay behind, also on a route
1079 without a response model, such as the list of a react-admin view.
1081 Restly copies the values of an instance of a subclass of ``cls``, and of
1082 an instance of the view's schema for a response class that
1083 :func:`derive_schema_response` built. It does not validate them again:
1084 they are already valid, and a validator that changes a value would
1085 change it twice.
1087 Any other model is validated into ``cls`` from its attributes. A response
1088 class that a view sets itself can have other fields and types, so Restly
1089 cannot copy into it.
1091 A value that is not a model comes back as it is. So does an ORM object,
1092 also a SQLModel table row, which is a model too: the view serializes it
1093 as an ORM object.
1094 """
1095 if type(value) is cls or not isinstance(value, pydantic.BaseModel):
1096 return value
1097 if is_mapped_instance(value): 1097 ↛ 1098line 1097 didn't jump to line 1098 because the condition on line 1097 was never true
1098 return value
1099 schema = cls.__dict__.get("__restly_response_of__")
1100 if not isinstance(value, cls) and (schema is None or not isinstance(value, schema)):
1101 return _response_adapter(cls).validate_python(
1102 value, from_attributes=True, by_alias=False, by_name=True
1103 )
1104 fields = cls.model_fields
1105 values = {name: item for name, item in value.__dict__.items() if name in fields}
1106 writeonly_keys = getattr(cls, "__restly_writeonly_keys__", frozenset())
1107 extra = {
1108 key: item
1109 for key, item in (value.__pydantic_extra__ or {}).items()
1110 if key not in writeonly_keys
1111 }
1112 copy = cls.model_construct(
1113 _fields_set=value.model_fields_set & fields.keys(), **extra, **values
1114 )
1115 if value.__pydantic_private__ is not None: 1115 ↛ 1116line 1115 didn't jump to line 1116 because the condition on line 1115 was never true
1116 object.__setattr__(
1117 copy, "__pydantic_private__", dict(value.__pydantic_private__)
1118 )
1119 return copy
1122@functools.cache
1123def _response_adapter(
1124 schema_response: type[pydantic.BaseModel],
1125) -> pydantic.TypeAdapter[pydantic.BaseModel]:
1126 """Validates into a response class. Through an adapter, not
1127 ``model_validate``: a schema may narrow that classmethod's signature
1128 (SQLModel drops ``by_alias`` and ``by_name``)."""
1129 return pydantic.TypeAdapter(schema_response)
1132_SchemaT = TypeVar("_SchemaT", bound=pydantic.BaseModel)
1135def derive_schema_response(schema: type[_SchemaT]) -> type[_SchemaT]:
1136 """Build the response class of a view's schema: the schema without its
1137 ``WriteOnly`` fields, named ``<Resource>Response``.
1139 A view without its own
1140 :attr:`~fastapi_restly.views.BaseRestView.schema_response` gets this
1141 class. Restly builds it also when the schema has no ``WriteOnly`` fields,
1142 so OpenAPI shows ``UserResponse`` and not the view's schema.
1144 The class is a subclass of ``schema``. Each call with the same schema
1145 returns the same class, so a custom route can name the class that the
1146 view uses::
1148 UserResponse = fr.schemas.derive_schema_response(UserSchema)
1151 class UserView(fr.AsyncRestView):
1152 schema = UserSchema
1154 @fr.get("/me", response_model=UserResponse)
1155 async def me(self): ...
1157 :param schema: The view's schema.
1158 :returns: The response class.
1159 """
1160 return cast(type[_SchemaT], _derive_schema_response(schema))
1163@functools.cache
1164def _derive_schema_response(
1165 schema: type[pydantic.BaseModel],
1166) -> type[pydantic.BaseModel]:
1167 """Cached, so a view, a custom route and every request use the same
1168 class. :func:`_as_response` turns an instance of the schema into an
1169 instance of this class."""
1170 cls = type(
1171 _schema_role_name(schema, "Response"),
1172 (_OmitWriteOnlyMixin, schema),
1173 {
1174 "__module__": schema.__module__,
1175 "__doc__": schema.__doc__,
1176 "__restly_response_of__": schema,
1177 },
1178 )
1179 # A recursive schema refers to itself by name, like list["NodeSchema"].
1180 # Pydantic finds a class's own name, but not the name of a parent class
1181 # that was defined in a function. So the response class gets that name.
1182 namespace = {**(schema.__pydantic_parent_namespace__ or {})}
1183 namespace[schema.__name__] = schema
1184 cls.model_rebuild(force=True, raise_errors=False, _types_namespace=namespace)
1185 return cls
1188def derive_schema_list_response(
1189 schema_response: type[pydantic.BaseModel],
1190 *,
1191 pagination: NumberedPagination | NoPagination | None = _DEFAULT_PAGINATION,
1192) -> type[pydantic.BaseModel]:
1193 """Build the list response class of a view: the envelope of its
1194 pagination, filled with its response class, named
1195 ``<Resource>ListResponse``.
1197 ``GET /`` of a view returns this class, except on a react-admin view,
1198 which returns a plain list. A custom list route names it, so OpenAPI
1199 shows one type for both routes::
1201 UserResponse = fr.schemas.derive_schema_response(UserSchema)
1202 UserListResponse = fr.schemas.derive_schema_list_response(
1203 UserResponse, pagination=AppView.pagination
1204 )
1207 class UserView(AppView):
1208 schema = UserSchema
1210 @fr.get("/inactive", response_model=UserListResponse)
1211 async def inactive(self, list_params): ...
1213 Each call with the same response class and envelope returns the same
1214 class. Only the envelope of the pagination matters: two paginations with
1215 the same envelope give the same class. A view has no attribute for this
1216 class. To change it, change the view's ``schema_response`` or the
1217 envelope of its ``pagination``.
1219 :param schema_response: The view's response class, such as the one from
1220 :func:`derive_schema_response`.
1221 :param pagination: The view's pagination. ``None`` is a view without
1222 pagination, as on the view. The default is the default pagination.
1223 :returns: The list response class.
1224 """
1225 envelope = Envelope if pagination is None else pagination.envelope
1226 return _named_list_envelope(envelope, schema_response)
1229@functools.cache
1230def _named_list_envelope(
1231 envelope: type[pydantic.BaseModel], schema_response: type[pydantic.BaseModel]
1232) -> type[pydantic.BaseModel]:
1233 """Cached, so the route and every request use the same class, and two
1234 views with the same response class and envelope share it.
1236 It is a subclass, so OpenAPI shows ``UserListResponse`` and not
1237 ``PaginatedEnvelope_UserResponse_``.
1238 """
1239 filled: Any = envelope[schema_response] # type: ignore[index]
1240 return type(
1241 _schema_role_name(schema_response, "ListResponse"),
1242 (filled,),
1243 {
1244 "__module__": schema_response.__module__,
1245 # For the warning about two classes with one name.
1246 "__restly_list_response_of__": (envelope, schema_response),
1247 },
1248 )
1251def rebase_with_model_config(
1252 base: tuple[type, ...], model_cls: type[pydantic.BaseModel]
1253) -> type[pydantic.BaseModel]:
1254 def class_body(ns: dict[str, Any]) -> None:
1255 ns["model_config"] = model_cls.model_config.copy()
1257 return types.new_class(
1258 f"{model_cls.__name__}ModelConfig", base, exec_body=class_body
1259 )
1262def create_model_with_optional_fields(
1263 model_cls: type[pydantic.BaseModel], *, omit: Iterable[str] = ()
1264) -> type[pydantic.BaseModel]:
1265 """
1266 Create a subclass of the given pydantic model class with a new name.
1267 Read-only fields and the fields named in ``omit`` are removed, and all
1268 writable fields are made optional with None as default.
1269 """
1270 new_model_name = _schema_role_name(model_cls, "Update")
1271 new_doc = (
1272 model_cls.__doc__ or ""
1273 ) + "\nRead-only fields have been removed and all fields are optional."
1275 # Create a subclass that mixes in both OmitReadOnlyMixin and PatchMixin
1276 new_model_cls = type(
1277 new_model_name,
1278 (PatchMixin, OmitReadOnlyMixin, model_cls),
1279 {
1280 "__module__": model_cls.__module__,
1281 "__doc__": new_doc,
1282 "__restly_omit_fields__": frozenset(omit),
1283 },
1284 )
1286 return new_model_cls
1289class PatchMixin(pydantic.BaseModel):
1290 """
1291 A mixin for pydantic classes that makes all fields optional and replaces defaults
1292 with None.
1294 Implementation note: like :class:`OmitReadOnlyMixin` this mutates
1295 ``cls.model_fields`` (specifically ``FieldInfo.default`` and
1296 ``FieldInfo.annotation``) and then calls ``model_rebuild(force=True)``.
1297 This relies on pydantic v2 keeping ``FieldInfo`` mutable; verified on the
1298 pinned ``pydantic>=2.11.7`` minimum and exercised by
1299 ``tests/test_pydantic_model_fields_mutation.py``.
1300 """
1302 @classmethod
1303 def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
1304 super().__pydantic_init_subclass__(**kwargs)
1306 for field in cls.model_fields.values():
1307 field.default = None
1308 # A field cannot have both. FastAPI builds the field again from
1309 # its attributes for OpenAPI, which raises with both.
1310 field.default_factory = None
1311 # Only wrap if not already Optional, to avoid Optional[Optional[T]]
1312 annotation = field.annotation
1313 if isinstance(annotation, types.UnionType):
1314 # Python 3.10+ `X | Y` syntax - check if None is already a member.
1315 # Convert to typing.Optional form so FieldInfo.annotation stays compatible.
1316 union_args = get_args(annotation)
1317 if type(None) not in union_args:
1318 non_none = [a for a in union_args if a is not type(None)]
1319 inner = (
1320 non_none[0] if len(non_none) == 1 else Union[tuple(non_none)]
1321 )
1322 field.annotation = Optional[inner] # type: ignore[assignment]
1323 else:
1324 origin = getattr(annotation, "__origin__", None)
1325 if origin is not Union or type(None) not in get_args(annotation):
1326 field.annotation = Optional[annotation] # type: ignore[assignment]
1328 cls.model_rebuild(force=True)
1331def getattrs(obj: Any, *attrs: str, default: Any = None) -> Any:
1332 """
1333 Try access a chain of attributes and return the default if any of the attrs is not defined.
1334 """
1335 for attr in attrs:
1336 if not hasattr(obj, attr):
1337 return default
1338 obj = getattr(obj, attr)
1339 return obj
1342def set_schema_title(schema_cls: type[pydantic.BaseModel]) -> None:
1343 """Set the title of a schema class to its name.
1344 This is used to make the schema title match the model name in the OpenAPI schema.
1345 """
1346 schema_cls.model_config["title"] = schema_cls.__name__
1349def get_writable_inputs(schema_obj: pydantic.BaseModel) -> dict[str, Any]:
1350 """
1351 Return a dictionary of field_name: value pairs for writable input fields.
1353 Filters out:
1354 - ReadOnly fields
1355 - fields not provided with input (using Pydantic model_fields_set)
1357 :param schema_obj: The schema object to extract writable fields from. Its
1358 own schema decides which fields are ``ReadOnly``.
1359 :returns: Dictionary mapping field names to their values, for writable
1360 input fields only.
1361 """
1362 schema_cls = type(schema_obj)
1363 updated_fields: dict[str, Any] = {}
1364 for field_name, value in schema_obj:
1365 if field_name not in schema_obj.model_fields_set:
1366 continue
1367 # Skip readonly fields
1368 if is_readonly_field(schema_cls, field_name):
1369 continue
1370 updated_fields[field_name] = value
1372 return updated_fields