Coverage for fastapi_restly / views / _base.py: 94%
1196 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-09 18:41 +0000
1"""
2This module provides a framework for class-based views on SQLAlchemy models.
4View class:
5This class is used to create a collection of endpoints that share an
6APIRouter (created when calling `include_view()`) and dependencies
7as class attributes. It uses the same mechanics as the class based
8view decorator from fastapi-utils.
9(https://fastapi-utils.davidmontague.xyz/user-guide/class-based-views/)
11AsyncRestView:
12Provides default reading and writing functions on the database using
13SQLAlchemy models.
14"""
16import dataclasses
17import functools
18import inspect
19import re
20import sys
21import types
22import warnings
23from collections.abc import Mapping, Set
24from enum import Enum
25from typing import (
26 TYPE_CHECKING,
27 Annotated,
28 Any,
29 Callable,
30 ClassVar,
31 Generic,
32 Iterable,
33 Iterator,
34 Protocol,
35 Sequence,
36 Union,
37 cast,
38 get_args,
39 get_origin,
40 get_type_hints,
41 overload,
42)
43from uuid import UUID
45import fastapi
46import pydantic
47from fastapi import BackgroundTasks, Request, Response, WebSocket
48from fastapi.datastructures import Default
49from fastapi.exceptions import RequestValidationError
50from fastapi.openapi.models import APIKeyIn
51from fastapi.params import Body as _BodyMarker
52from fastapi.params import Depends as _DependsMarker
53from fastapi.params import Param as _ParamMarker
54from fastapi.security.api_key import APIKeyBase
55from pydantic.fields import FieldInfo
56from sqlalchemy import JSON as _JSONType
57from sqlalchemy import ColumnElement, Select
58from sqlalchemy import inspect as sa_inspect
59from sqlalchemy import select as sa_select
60from sqlalchemy.orm import selectinload
61from sqlalchemy.orm.attributes import del_attribute
62from starlette.datastructures import QueryParams
63from typing_extensions import TypeVar
65from .._exception_handlers import register_default_exception_handlers
67if sys.version_info >= (3, 14): 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true
68 import annotationlib
69from ..clauses import (
70 UNSCOPED,
71 Unscoped,
72 WhereClause,
73 all_of,
74 apply_clauses,
75 where_clause,
76)
77from ..clauses._scopes import _default_scope
79#: A per-read scope: ``None`` for the view's own, a clause that replaces it
80#: for that read, or ``fr.clauses.UNSCOPED``.
81ReadScope = WhereClause | Unscoped | None
83# A per-read narrowing filter: a SQLAlchemy boolean expression, or anything
84# all_of takes. Private: only the final list handler accepts one, so no
85# override spells it.
86_ReadWhere = ColumnElement[bool] | WhereClause | Unscoped | None
87from .._mapping import is_mapped_class, is_mapped_instance
88from .._pagination import (
89 _DEFAULT_PAGINATION,
90 DEFAULT_PAGE_SIZE,
91 NoPagination,
92 NumberedPagination,
93 _build_envelope,
94 _numbered_page_info,
95 _unpaginated_page_info,
96)
97from .._routes import matched_route
98from ..db._globals import _fr_globals
99from ..exc import RestlyConfigurationError, RestlyMisuseWarning
100from ..objects import snapshot as _object_snapshot
101from ..query import derive_schema_list_params
102from ..query._impl import _UNKNOWN, _created_with, _query_settings
103from ..schemas import (
104 BaseSchema,
105 IDSchema,
106 derive_schema_list_response,
107 derive_schema_response,
108)
109from ..schemas._base import (
110 _as_response,
111 _model_id_type,
112 _reject_buried_markers,
113 _response_adapter,
114 _unwrap_optional_annotation,
115 create_model_with_optional_fields,
116 create_model_without_read_only_fields,
117 get_writable_inputs,
118 is_readonly_field,
119 is_reference_field,
120 reference_origin_and_target,
121)
122from ..schemas._generator import derive_schema
123from ._openapi import _register_for_resource_ref
125# Unbound: a mapped class need not subclass DeclarativeBase (a SQLModel table,
126# an imperative mapping). ``BaseRestView`` checks the mapping at class definition.
127ModelT = TypeVar("ModelT", default=Any)
128ResponseSchemaT = TypeVar(
129 "ResponseSchemaT", bound=pydantic.BaseModel, default=BaseSchema
130)
131CreateSchemaT = TypeVar(
132 "CreateSchemaT", bound=pydantic.BaseModel, default=pydantic.BaseModel
133)
134UpdateSchemaT = TypeVar(
135 "UpdateSchemaT", bound=pydantic.BaseModel, default=pydantic.BaseModel
136)
137IdT = TypeVar("IdT", default=int)
140@dataclasses.dataclass(frozen=True)
141class ListResult(Generic[ModelT]):
142 """Result returned by ``get_many`` before HTTP response formatting.
144 ``total_count`` is ``None`` for a list without pagination, which does not
145 run the count query. ``list_params`` holds the list params that
146 ``get_many`` received, so :meth:`BaseRestView.to_list_response` can read
147 the page from them.
148 """
150 objects: Sequence[ModelT]
151 total_count: int | None = None
152 list_params: Any = None
155class ViewRoute(str, Enum):
156 """Default CRUD route names that can be referenced by view options.
158 Values are the endpoint method names so ``exclude_routes`` can drop them.
159 """
161 GET_MANY = "get_many_endpoint"
162 GET_ONE = "get_one_endpoint"
163 CREATE = "create_endpoint"
164 UPDATE = "update_endpoint"
165 DELETE = "delete_endpoint"
168class ResponseShape(str, Enum):
169 """The wire shape an endpoint method asks :meth:`BaseRestView.to_response` to
170 produce.
172 This is separate from write-action names such as ``"publish"``. Endpoint
173 methods choose one of these three response shapes; custom actions remain an
174 open string namespace.
175 """
177 SINGLE = "single" # one serialized object
178 LIST = "list" # a ListResult -> the list envelope
179 EMPTY = "empty" # 204 No Content
182class Action:
183 """Canonical CRUD action names passed to ``authorize`` / ``before_action_commit``
184 / ``after_action_commit``.
186 This is a constants class, not an ``Enum``: custom actions and mixins add
187 their own names. Use constants for typo checking at import time.
188 """
190 GET_MANY = "get_many"
191 GET_ONE = "get_one"
192 CREATE = "create"
193 UPDATE = "update"
194 DELETE = "delete"
197def _accepts_init_kwarg(model_cls: type, attr_name: str) -> bool:
198 """Return True if attr_name can be passed as a keyword argument to model_cls.__init__.
200 Non-dataclass models (DeclarativeBase subclasses using mapped_column) accept all
201 kwargs. Dataclass-based models may have fields with init=False, in which case
202 passing the attribute to __init__ raises TypeError.
203 """
204 if not dataclasses.is_dataclass(model_cls):
205 return True
206 dc_fields = {f.name: f for f in dataclasses.fields(model_cls)}
207 return attr_name not in dc_fields or dc_fields[attr_name].init
210def _requires_init_kwarg(model_cls: type, attr_name: str) -> bool:
211 if not dataclasses.is_dataclass(model_cls):
212 return False
213 dc_fields = {f.name: f for f in dataclasses.fields(model_cls)}
214 field = dc_fields.get(attr_name)
215 if field is None or not field.init:
216 return False
217 return (
218 field.default is dataclasses.MISSING
219 and field.default_factory is dataclasses.MISSING
220 )
223@dataclasses.dataclass
224class _CreatePlan:
225 kwargs: dict[str, Any]
226 post_assignments: dict[str, Any]
227 # Many-to-one relationships the input did not supply.
228 unsupplied_relationships: list[str] = dataclasses.field(default_factory=list)
231class _HasID(Protocol):
232 """Anything with an ``id`` attribute. By framework convention, primary
233 keys are named ``id``; ``IDBase`` formalizes this but isn't required."""
235 id: Any
238def _has_model_attr(model_cls: type[Any], attr_name: str) -> bool:
239 return hasattr(model_cls, attr_name)
242def _get_relationship_property(model_cls: type[Any], relation_name: str) -> Any | None:
243 try:
244 mapper = sa_inspect(model_cls)
245 except Exception:
246 return None
247 return mapper.relationships.get(relation_name)
250def _column_attr_name(mapper: Any, column: Any) -> str | None:
251 """Map a Column object back to its mapped attribute name.
253 ``Mapper.columns`` is keyed by the mapped attribute name, but a Column's own
254 ``key`` is the DB column name; the two differ when a column is declared
255 ``mapped_column("db_name", ...)``. Reference routing works in attribute-name
256 space (kwargs and ``setattr``), so callers need the attribute name, not the
257 column key. Compared by identity because ``Column.__eq__`` builds a SQL
258 expression rather than a boolean.
259 """
260 for attr_name, mapped_column in mapper.columns.items(): 260 ↛ 263line 260 didn't jump to line 263 because the loop on line 260 didn't complete
261 if mapped_column is column:
262 return attr_name
263 return None
266def _get_unambiguous_local_fk_name(
267 model_cls: type[Any], relation_name: str
268) -> str | None:
269 relationship_property = _get_relationship_property(model_cls, relation_name)
270 if relationship_property is None: 270 ↛ 271line 270 didn't jump to line 271 because the condition on line 270 was never true
271 return None
273 if getattr(relationship_property.direction, "name", None) != "MANYTOONE": 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true
274 return None
276 local_columns = list(relationship_property.local_columns)
277 if len(local_columns) != 1:
278 column_names = ", ".join(column.key for column in local_columns) or "<none>"
279 raise ValueError(
280 f"Cannot infer a single local FK for relationship "
281 f"{model_cls.__name__}.{relation_name}; found {column_names}. "
282 "Use an explicit custom handler for this relationship."
283 )
284 # Return the mapped attribute name (what kwargs/setattr need), which differs
285 # from the column's DB-name ``key`` under ``mapped_column("db_name", ...)``.
286 try:
287 mapper = sa_inspect(model_cls)
288 except Exception:
289 return local_columns[0].key
290 return _column_attr_name(mapper, local_columns[0]) or local_columns[0].key
293def _many_to_one_names(model_cls: type[Any]) -> list[str]:
294 try:
295 mapper = sa_inspect(model_cls)
296 except Exception:
297 return []
298 return [
299 relationship_property.key
300 for relationship_property in mapper.relationships
301 if getattr(relationship_property.direction, "name", None) == "MANYTOONE"
302 ]
305def _relationship_name_for_fk(model_cls: type[Any], fk_attr_name: str) -> str | None:
306 """Reverse of :func:`_get_unambiguous_local_fk_name`.
308 Given a scalar FK column's attribute name, return the single many-to-one
309 relationship that uses it as its local column, or ``None`` when there is no
310 such relationship or more than one (ambiguous, so no pairing is guessed).
311 This is how a FK reference field finds its partner relationship without
312 relying on the ``<relation>_id`` naming convention.
313 """
314 try:
315 mapper = sa_inspect(model_cls)
316 except Exception:
317 return None
318 fk_column = mapper.columns.get(fk_attr_name)
319 if fk_column is None: 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 return None
321 # Match by Column identity, not by ``.key``: the column's key is the DB name,
322 # which differs from ``fk_attr_name`` under ``mapped_column("db_name", ...)``.
323 # (``is`` also avoids ``Column.__eq__``'s SQL-expression overload.)
324 matches = [
325 relationship_property.key
326 for relationship_property in mapper.relationships
327 if getattr(relationship_property.direction, "name", None) == "MANYTOONE"
328 and any(column is fk_column for column in relationship_property.local_columns)
329 ]
330 return matches[0] if len(matches) == 1 else None
333def _is_mapped_column(model_cls: type[Any], field_name: str) -> bool:
334 """True if ``field_name`` maps to a (scalar) column on ``model_cls``.
336 This is the target for writing a resolved reference's raw id. Decided by the
337 mapper rather than the field name, so a FK column works under any name, not
338 only ``<relation>_id``.
339 """
340 try:
341 mapper = sa_inspect(model_cls)
342 except Exception:
343 return False
344 return field_name in mapper.columns
347def _is_json_column(model_cls: type[Any], field_name: str) -> bool:
348 """True if ``field_name`` maps to a plain ``JSON`` column (``JSONB`` too).
350 A ``TypeDecorator`` over ``JSON`` is not a ``JSON`` instance and so is not
351 one: it has its own bind processor, which may well want the object as it
352 stands.
353 """
354 try:
355 mapper = sa_inspect(model_cls)
356 except Exception:
357 return False
358 column = mapper.columns.get(field_name)
359 return column is not None and isinstance(column.type, _JSONType)
362def _json_ready(model_cls: type[Any], field_name: str, value: Any) -> Any:
363 """Dump pydantic models on their way into a JSON column.
365 A schema field typed as a nested model validates to a model instance, and
366 a JSON column binds through ``json.dumps``, which cannot take one. Restly
367 owns the schema-to-model translation, so it does the dump here rather than
368 letting the driver fail at flush with the model in the bind parameters.
369 """
370 if isinstance(value, pydantic.BaseModel):
371 if not _is_json_column(model_cls, field_name):
372 return value
373 _reject_excluded_json_fields(value, f"{model_cls.__name__}.{field_name}")
374 return value.model_dump(mode="json")
375 if isinstance(value, list | tuple) and any(
376 isinstance(item, pydantic.BaseModel) for item in value
377 ):
378 if not _is_json_column(model_cls, field_name): 378 ↛ 379line 378 didn't jump to line 379 because the condition on line 378 was never true
379 return value
380 _reject_excluded_json_fields(value, f"{model_cls.__name__}.{field_name}")
381 return [
382 item.model_dump(mode="json")
383 if isinstance(item, pydantic.BaseModel)
384 else item
385 for item in value
386 ]
387 return value
390def _reject_excluded_json_fields(value: Any, column: str) -> None:
391 """Reject exclusions and document values that cannot be inspected safely."""
392 pending = [value]
393 seen: set[int] = set()
394 while pending:
395 item = pending.pop()
396 if dataclasses.is_dataclass(item) or isinstance(item, Iterator):
397 raise RestlyConfigurationError(
398 f"{column}: documents containing dataclasses or iterators cannot "
399 "be stored automatically in a plain JSON column. Use a SQLAlchemy "
400 "TypeDecorator to define their storage representation."
401 )
402 if isinstance(item, str | bytes | bytearray) or not isinstance(
403 item, (pydantic.BaseModel, Mapping, Sequence, Set, Enum)
404 ):
405 continue
406 if id(item) in seen: 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true
407 continue
408 seen.add(id(item))
409 if isinstance(item, pydantic.BaseModel):
410 for name, field in type(item).model_fields.items():
411 if field.exclude:
412 raise RestlyConfigurationError(
413 f"{column}: {type(item).__name__}.{name} is excluded from "
414 "serialization. Documents with WriteOnly or "
415 "Field(exclude=True) fields cannot be stored automatically "
416 "in a plain JSON column. Use a SQLAlchemy TypeDecorator "
417 "to define their storage representation."
418 )
419 pending.append(getattr(item, name, None))
420 if item.model_extra:
421 pending.extend(item.model_extra.values())
422 elif isinstance(item, Mapping):
423 pending.extend(item.keys())
424 pending.extend(item.values())
425 elif isinstance(item, Enum):
426 pending.append(item.value)
427 else:
428 pending.extend(item)
431def _add_assignment(target: dict[str, Any], field_name: str | None, value: Any) -> None:
432 if field_name: 432 ↛ exitline 432 didn't return from function '_add_assignment' because the condition on line 432 was always true
433 target[field_name] = value
436_EXPLICIT_NULL_REF = object()
439def _reference_identity(value: Any) -> tuple[type[Any] | None, Any] | object | None:
440 if value is None:
441 return _EXPLICIT_NULL_REF
442 if is_mapped_instance(value): 442 ↛ 444line 442 didn't jump to line 444 because the condition on line 442 was always true
443 return type(value), getattr(value, "id", None)
444 if isinstance(value, IDSchema):
445 sql_model = value.get_sql_model_annotation()
446 return sql_model, value.id
447 return None
450def _reference_identity_detail(identity: object) -> Any:
451 if identity is _EXPLICIT_NULL_REF:
452 return None
453 if isinstance(identity, tuple) and len(identity) == 2: 453 ↛ 455line 453 didn't jump to line 455 because the condition on line 453 was always true
454 return identity[1]
455 return identity
458def validate_resolved_reference_consistency(
459 model_cls: type[Any],
460 schema_obj: pydantic.BaseModel,
461 resolved: dict[str, Any] | None = None,
462) -> None:
463 """Validate explicitly supplied FK and relationship fields agree.
465 IDRef/IDSchema resolution looks model-aware references up as ORM objects
466 (in ``resolved``) before object construction/update. If the client supplied
467 both ``author_id`` and ``author`` independently, they must refer to the same
468 row. ``resolved`` is the ``{field: object}`` mapping from the resolver; a
469 field absent from it keeps its (unresolved) value on ``schema_obj``.
470 """
471 schema_cls = type(schema_obj)
472 resolved = resolved or {}
474 for fk_field in schema_obj.model_fields_set:
475 # Start from the FK-column side of a reference pair and derive its
476 # partner relationship from the mapper; relationship-named fields are
477 # reached as the partner, not iterated here.
478 if not is_reference_field(schema_cls, fk_field):
479 continue
480 if _get_relationship_property(model_cls, fk_field) is not None:
481 continue
483 relation_field = _relationship_name_for_fk(model_cls, fk_field)
484 if (
485 relation_field is None
486 or relation_field not in schema_obj.model_fields_set
487 or not is_reference_field(schema_cls, relation_field)
488 ):
489 continue
491 fk_identity = _reference_identity(
492 resolved.get(fk_field, getattr(schema_obj, fk_field, None))
493 )
494 relation_identity = _reference_identity(
495 resolved.get(relation_field, getattr(schema_obj, relation_field, None))
496 )
497 if fk_identity is None or relation_identity is None: 497 ↛ 498line 497 didn't jump to line 498 because the condition on line 497 was never true
498 continue
500 if fk_identity == relation_identity:
501 continue
503 raise fastapi.HTTPException(
504 status_code=422,
505 detail=(
506 f"Conflicting references for {fk_field} and {relation_field}: "
507 f"{_reference_identity_detail(fk_identity)!r} != "
508 f"{_reference_identity_detail(relation_identity)!r}"
509 ),
510 )
513def iter_creatable_fields(schema_obj: pydantic.BaseModel) -> Iterator[tuple[str, Any]]:
514 """Iterate over (field_name, value) pairs that should be used to construct a new
515 ORM object from ``schema_obj``.
517 Fields marked as ``ReadOnly`` are skipped. Unlike :func:`get_writable_inputs`,
518 this also includes fields that were not explicitly provided, so that
519 schema-level defaults end up on the new object.
520 """
521 schema_cls = type(schema_obj)
522 for field_name, value in schema_obj:
523 if is_readonly_field(schema_cls, field_name):
524 continue
525 yield field_name, value
528def _add_null_reference_to_create_plan(
529 plan: _CreatePlan, model_cls: type[Any], field_name: str
530) -> None:
531 """Write a null reference (explicit ``post=None``, or an omitted field
532 defaulting to None) into the plan.
534 Unlike a resolved reference, a null writes only the field's own slot, plus
535 the partner FK column when the dataclass ``__init__`` requires it. It never
536 mirrors further, and never overwrites a slot another field already planned:
537 schemas may declare both names of an FK/relationship pair as reference
538 fields, and the unset sibling's default None must not clobber the side the
539 client supplied. (Later fields overwrite a planned None naturally, so the
540 ``setdefault`` guards are order-independent. A relationship the input did
541 not supply is left to :func:`build_create_plan`, which passes None only
542 where ``__init__`` requires it and has it unset after construction.)
543 """
544 relationship = _get_relationship_property(model_cls, field_name)
545 if relationship is None and not _is_mapped_column(model_cls, field_name): 545 ↛ 546line 545 didn't jump to line 546 because the condition on line 545 was never true
546 return # unmapped reference name; nothing to write
548 if _accepts_init_kwarg(model_cls, field_name):
549 plan.kwargs.setdefault(field_name, None)
550 elif _has_model_attr(model_cls, field_name): 550 ↛ 553line 550 didn't jump to line 553 because the condition on line 550 was always true
551 plan.post_assignments.setdefault(field_name, None)
553 if relationship is None:
554 return
556 # A required-init partner FK column must be constructed even for a null
557 # reference, or ``__init__`` rejects the missing kwarg. A composite-FK
558 # relationship has no single partner column — but a null also has no id to
559 # route, so there is nothing to infer; skip instead of raising.
560 try:
561 fk_name = _get_unambiguous_local_fk_name(model_cls, field_name)
562 except ValueError:
563 return
564 if fk_name and _requires_init_kwarg(model_cls, fk_name):
565 plan.kwargs.setdefault(fk_name, None)
568def _add_resolved_reference_to_create_plan(
569 plan: _CreatePlan, model_cls: type[Any], field_name: str, value: Any
570) -> None:
571 # ``cast`` only; ``.id`` is accessed lazily inside the branches so a
572 # composite-keyed relationship raises its descriptive ValueError (from
573 # the single-FK inference) before any ``.id`` attribute error.
574 ref = cast(_HasID, value)
575 # Route by what the field maps to on the model (mapper introspection), not
576 # by its name: a relationship gets the ORM object, a scalar FK column gets
577 # the row's id. The partner attribute is derived from the mapper so this
578 # works for any column name, not only the ``<relation>_id`` convention.
579 relationship = _get_relationship_property(model_cls, field_name)
580 if relationship is not None:
581 relation_name = field_name
582 fk_name = _get_unambiguous_local_fk_name(model_cls, relation_name)
584 # A required-init FK column must be constructed, not post-assigned, or
585 # the dataclass __init__ rejects the missing kwarg. Pass its id at
586 # construction: alongside the relationship object when that too is an
587 # init kwarg (consistent ids), or instead of it when it isn't.
588 if fk_name and _requires_init_kwarg(model_cls, fk_name):
589 plan.kwargs[fk_name] = ref.id
590 if _accepts_init_kwarg(model_cls, relation_name):
591 plan.kwargs[relation_name] = value
592 else:
593 plan.post_assignments[relation_name] = value
594 return
596 if _accepts_init_kwarg(model_cls, relation_name):
597 plan.kwargs[relation_name] = value
598 _add_assignment(plan.post_assignments, fk_name, ref.id)
599 return
601 if fk_name and _accepts_init_kwarg(model_cls, fk_name): 601 ↛ 602line 601 didn't jump to line 602 because the condition on line 601 was never true
602 plan.kwargs[fk_name] = ref.id
603 plan.post_assignments[relation_name] = value
604 return
606 plan.post_assignments[relation_name] = value
607 _add_assignment(plan.post_assignments, fk_name, ref.id)
608 return
610 if not _is_mapped_column(model_cls, field_name): 610 ↛ 614line 610 didn't jump to line 614 because the condition on line 610 was never true
611 # Reference field that maps to neither a relationship nor a column;
612 # there is nothing to write (matches the relationship branch's drop of
613 # unmapped names).
614 return
616 fk_name = field_name
617 relation_name = _relationship_name_for_fk(model_cls, fk_name)
619 if relation_name is None:
620 if _accepts_init_kwarg(model_cls, fk_name): 620 ↛ 623line 620 didn't jump to line 623 because the condition on line 620 was always true
621 plan.kwargs[fk_name] = ref.id
622 else:
623 plan.post_assignments[fk_name] = ref.id
624 return
626 # A partner relationship exists; keep the FK column and the relationship
627 # object consistent, honoring dataclass init requirements on either side.
628 accepts_relation = _accepts_init_kwarg(model_cls, relation_name)
629 relation_required = _requires_init_kwarg(model_cls, relation_name)
631 if (
632 _requires_init_kwarg(model_cls, fk_name)
633 and accepts_relation
634 and relation_required
635 ):
636 plan.kwargs[fk_name] = ref.id
637 plan.kwargs[relation_name] = value
638 return
640 if accepts_relation and relation_required: 640 ↛ 641line 640 didn't jump to line 641 because the condition on line 640 was never true
641 plan.kwargs[relation_name] = value
642 plan.post_assignments[fk_name] = ref.id
643 return
645 if _accepts_init_kwarg(model_cls, fk_name):
646 plan.kwargs[fk_name] = ref.id
647 plan.post_assignments[relation_name] = value
648 return
650 if accepts_relation:
651 plan.kwargs[relation_name] = value
652 plan.post_assignments[fk_name] = ref.id
653 return
655 plan.post_assignments[fk_name] = ref.id
656 plan.post_assignments[relation_name] = value
659def build_create_plan(
660 model_cls: type[Any],
661 schema_obj: pydantic.BaseModel,
662 resolved: dict[str, Any] | None = None,
663) -> _CreatePlan:
664 """Translate ``schema_obj`` fields into kwargs for ``model_cls(**kwargs)``.
666 Shared by sync and async ``make_new_object``. ``resolved`` is the
667 ``{field: object_or_list}`` mapping returned by the IDSchema resolver (sync
668 vs async); a resolved reference field uses that ORM value instead of the
669 wire-shaped ``IDRef`` still on ``schema_obj``.
670 """
671 schema_cls = type(schema_obj)
672 resolved = resolved or {}
674 plan = _CreatePlan(kwargs={}, post_assignments={})
675 for field_name, value in iter_creatable_fields(schema_obj):
676 if field_name in resolved:
677 value = resolved[field_name]
678 if isinstance(value, IDSchema) and _is_mapped_column(model_cls, field_name):
679 if _accepts_init_kwarg(model_cls, field_name): 679 ↛ 682line 679 didn't jump to line 682 because the condition on line 679 was always true
680 plan.kwargs[field_name] = value.id
681 else:
682 plan.post_assignments[field_name] = value.id
683 continue
684 if value is None and is_reference_field(schema_cls, field_name):
685 _add_null_reference_to_create_plan(plan, model_cls, field_name)
686 continue
687 if is_mapped_instance(value) and is_reference_field(schema_cls, field_name):
688 _add_resolved_reference_to_create_plan(plan, model_cls, field_name, value)
689 continue
691 value = _json_ready(model_cls, field_name, value)
692 if _accepts_init_kwarg(model_cls, field_name):
693 plan.kwargs[field_name] = value
694 elif _has_model_attr(model_cls, field_name): 694 ↛ 675line 694 didn't jump to line 675 because the condition on line 694 was always true
695 plan.post_assignments[field_name] = value
697 plan.unsupplied_relationships = [
698 name
699 for name in _many_to_one_names(model_cls)
700 if name not in schema_obj.model_fields_set
701 or is_readonly_field(schema_cls, name)
702 ]
703 for name in plan.unsupplied_relationships:
704 # __init__ demands the argument; the None is unset after construction.
705 if _requires_init_kwarg(model_cls, name):
706 plan.kwargs.setdefault(name, None)
707 return plan
710def apply_create_assignments(obj: Any, plan: _CreatePlan) -> None:
711 """Finish an object built from ``plan.kwargs``.
713 Sets the attributes the constructor cannot take, then unsets each
714 many-to-one relationship the input did not supply that holds ``None``.
715 That ``None`` is a default: the dataclass ``__init__`` assigns
716 ``relationship(default=None)``, and the plan passes ``None`` for an
717 omitted reference field or a required relationship argument. SQLAlchemy's
718 flush would copy it over the foreign key, discarding an id the input did
719 supply. Unset, the relationship takes no part in the flush, as SQLAlchemy
720 2.1 treats dataclass defaults. A ``null`` the client sent stays.
721 """
722 for field_name, value in plan.post_assignments.items():
723 setattr(obj, field_name, value)
724 loaded = sa_inspect(obj).dict
725 for name in plan.unsupplied_relationships:
726 if name in loaded and loaded[name] is None:
727 del_attribute(obj, name)
730def _apply_resolved_reference_update(obj: Any, field_name: str, value: Any) -> None:
731 ref = cast(_HasID, value)
732 model_cls = type(obj)
733 relationship = _get_relationship_property(model_cls, field_name)
734 if relationship is None:
735 # Scalar FK column: write the id, and mirror onto the partner
736 # relationship (derived from the mapper) when there is one.
737 if not _is_mapped_column(model_cls, field_name): 737 ↛ 738line 737 didn't jump to line 738 because the condition on line 737 was never true
738 return
739 setattr(obj, field_name, ref.id)
740 relation_name = _relationship_name_for_fk(model_cls, field_name)
741 if relation_name is not None:
742 setattr(obj, relation_name, value)
743 return
745 # Relationship field: assign the object, mirror onto the local FK if any.
746 setattr(obj, field_name, value)
747 fk_name = _get_unambiguous_local_fk_name(model_cls, field_name)
748 if fk_name: 748 ↛ exitline 748 didn't return from function '_apply_resolved_reference_update' because the condition on line 748 was always true
749 setattr(obj, fk_name, ref.id)
752def apply_update_to_object(
753 obj: Any, schema_obj: pydantic.BaseModel, resolved: dict[str, Any] | None = None
754) -> None:
755 """Apply writable inputs from ``schema_obj`` onto ``obj`` in place.
757 Shared by sync and async ``update_object``. ``resolved`` is the
758 ``{field: object_or_list}`` mapping returned by the IDSchema resolver (sync
759 vs async); a resolved reference field uses that ORM value instead of the
760 wire-shaped ``IDRef`` still on ``schema_obj``.
761 """
762 schema_cls = type(schema_obj)
763 resolved = resolved or {}
764 for field_name, value in get_writable_inputs(schema_obj).items():
765 if field_name in resolved:
766 value = resolved[field_name]
767 if isinstance(value, IDSchema) and _is_mapped_column(type(obj), field_name):
768 setattr(obj, field_name, value.id)
769 continue
770 if is_mapped_instance(value) and is_reference_field(schema_cls, field_name):
771 _apply_resolved_reference_update(obj, field_name, value)
772 continue
773 setattr(obj, field_name, _json_ready(type(obj), field_name, value))
776def _get_nested_schema_annotation(annotation: Any) -> type[pydantic.BaseModel] | None:
777 annotation = _unwrap_optional_annotation(annotation)
779 try:
780 if inspect.isclass(annotation) and issubclass(annotation, pydantic.BaseModel):
781 return annotation
782 except TypeError:
783 pass
785 origin = get_origin(annotation)
786 if origin is list: 786 ↛ 791line 786 didn't jump to line 791 because the condition on line 786 was always true
787 args = get_args(annotation)
788 if args: 788 ↛ 791line 788 didn't jump to line 791 because the condition on line 788 was always true
789 return _get_nested_schema_annotation(args[0])
791 return None
794def _generated_primary_key_fields(model_cls: type[Any]) -> frozenset[str]:
795 """The primary key attributes the server generates, never the client.
797 A derived create or update schema leaves these out even when the schema
798 does not mark them ReadOnly. Generated means an autoincrement column, a
799 column default, or a dataclass field outside ``__init__``. A key without
800 any of these is a natural key, and the client supplies it.
801 """
802 mapper = sa_inspect(model_cls)
803 autoincrement = getattr(mapper.local_table, "autoincrement_column", None)
804 not_in_init = (
805 {f.name for f in dataclasses.fields(model_cls) if not f.init}
806 if dataclasses.is_dataclass(model_cls)
807 else set()
808 )
809 generated = set()
810 for column in mapper.primary_key:
811 key = mapper.get_property_by_column(column).key
812 if (
813 column is autoincrement
814 or column.default is not None
815 or column.server_default is not None
816 or key in not_in_init
817 ):
818 generated.add(key)
819 return frozenset(generated)
822def _build_relationship_loader_options(
823 model_cls: type[Any],
824 schema_cls: type[pydantic.BaseModel],
825 seen: set[tuple[type[Any], type[pydantic.BaseModel]]] | None = None,
826) -> list[Any]:
827 if seen is None:
828 seen = set()
830 visit_key = (model_cls, schema_cls)
831 if visit_key in seen: 831 ↛ 832line 831 didn't jump to line 832 because the condition on line 831 was never true
832 return []
833 seen = seen | {visit_key}
835 mapper = sa_inspect(model_cls)
836 options: list[Any] = []
837 for field_name, field_info in schema_cls.model_fields.items():
838 if field_name not in mapper.relationships:
839 continue
841 relationship_prop = mapper.relationships[field_name]
842 loader = selectinload(getattr(model_cls, field_name))
843 nested_schema = _get_nested_schema_annotation(field_info.annotation)
845 if nested_schema is not None: 845 ↛ 852line 845 didn't jump to line 852 because the condition on line 845 was always true
846 child_options = _build_relationship_loader_options(
847 relationship_prop.mapper.class_, nested_schema, seen
848 )
849 if child_options:
850 loader = loader.options(*child_options)
852 options.append(loader)
854 return options
857def _relationship_reload_statement(obj: Any, options: list[Any]) -> Any:
858 """A primary-key SELECT that fills ``obj``'s unloaded relationships.
860 Deliberately *not* ``populate_existing``: that would expire every
861 relationship the options do not name and overwrite the ones they do, so a
862 value the caller had already put there would be replaced by a fresh read.
863 Without it the loaders fill only what is unloaded, which is exactly what
864 serialization would otherwise have lazy-loaded.
865 """
866 state = sa_inspect(obj)
867 if state.identity is None:
868 return None
869 mapper = state.mapper
870 conditions = [
871 column == value
872 for column, value in zip(mapper.primary_key, state.identity, strict=True)
873 ]
874 return sa_select(mapper.class_).where(*conditions).options(*options)
877def _schema_relationships_are_loaded(
878 obj: Any,
879 model_cls: type[Any],
880 schema_cls: type[pydantic.BaseModel],
881 seen: set[tuple[int, type[pydantic.BaseModel]]] | None = None,
882) -> bool:
883 """Is every relationship ``schema_cls`` reaches already loaded on ``obj``?
885 Walks the same model/schema pairing as
886 :func:`_build_relationship_loader_options`, so it answers exactly the
887 question "would serializing this object need IO?". Reads only the committed
888 attribute dict, never the descriptors, so asking cannot itself emit a query.
889 """
890 if seen is None:
891 seen = set()
892 # Dedup on (instance, schema), not the instance alone: the same object can
893 # be reached under two response sub-schemas that name different
894 # relationships, and each pairing must be checked. Mirrors the
895 # (model, schema) dedup in _build_relationship_loader_options.
896 key = (id(obj), schema_cls)
897 if key in seen:
898 return True
899 seen.add(key)
901 state = sa_inspect(obj, raiseerr=False)
902 if state is None: 902 ↛ 903line 902 didn't jump to line 903 because the condition on line 902 was never true
903 return True
905 mapper = sa_inspect(model_cls)
906 for field_name, field_info in schema_cls.model_fields.items():
907 if field_name not in mapper.relationships:
908 continue
909 if field_name in state.unloaded:
910 return False
912 nested_schema = _get_nested_schema_annotation(field_info.annotation)
913 if nested_schema is None: 913 ↛ 914line 913 didn't jump to line 914 because the condition on line 913 was never true
914 continue
916 value = state.dict.get(field_name)
917 if value is None:
918 continue
919 relationship_prop = mapper.relationships[field_name]
920 related = value if relationship_prop.uselist else [value]
921 for item in related:
922 if not _schema_relationships_are_loaded(
923 item, relationship_prop.mapper.class_, nested_schema, seen
924 ):
925 return False
927 return True
930class View:
931 """
932 Class-based view primitive for FastAPI.
934 Group related endpoints on a class, share dependencies and metadata via
935 class attributes, and let subclasses override individual handlers. Routes
936 are bound at :func:`include_view` time, not at class-definition time, so
937 subclassing works the way Python developers expect: override a method on
938 a subclass and the override is what runs.
940 Most users will subclass :class:`RestView` or :class:`AsyncRestView`,
941 which extend ``View`` with CRUD scaffolding. Use ``View`` directly for
942 grouped non-CRUD endpoints (auth flows, custom RPC routes, etc.).
943 """
945 #: The URL prefix of every route. Each class in the hierarchy that sets one
946 #: adds a segment, base first.
947 prefix: ClassVar[str]
948 tags: ClassVar[Iterable[str | Enum] | None] = None
949 #: FastAPI dependencies run for every route, without injecting a result.
950 #: Each class in the hierarchy adds its own, base first, so a subclass
951 #: cannot drop a base's guard. A subclass that lists a base's entry again
952 #: runs it where it lists it.
953 dependencies: ClassVar[Any] = None
954 #: OpenAPI responses documented on every route. Each class in the hierarchy
955 #: adds its own, and a subclass's entry for a status code wins.
956 responses: ClassVar[dict[int | str, dict[str, Any]]] = {}
957 #: FastAPI route keyword arguments keyed by endpoint method name or
958 #: ``ViewRoute``. These override decorator metadata without replacing the
959 #: endpoint method. A subclass inherits this mapping unless it declares
960 #: its own, which replaces it.
961 route_options: ClassVar[Mapping[str | ViewRoute, Mapping[str, Any]]] = {}
963 if TYPE_CHECKING:
964 # include_view replaces __init__: each injected attribute becomes a
965 # keyword, so code that builds a view by hand passes them by name.
966 def __init__(self, **dependencies: Any) -> None: ...
968 @classmethod
969 def before_include_view(cls) -> None:
970 """Run by :func:`include_view` once per class, before its routes are
971 registered. A no-op here; override to adjust route methods first."""
974V = TypeVar("V", bound=type[View])
977@overload
978def include_view(
979 parent_router: fastapi.APIRouter | fastapi.FastAPI, view_cls: V
980) -> V: ...
981@overload
982def include_view(
983 parent_router: fastapi.APIRouter | fastapi.FastAPI,
984) -> Callable[[V], V]: ...
987def include_view(
988 parent_router: fastapi.APIRouter | fastapi.FastAPI, view_cls: V | None = None
989) -> V | Callable[[V], V]:
990 """
991 Add a View class's routes to a FastAPI app or APIRouter.
993 Prefer the direct call form from your app/router composition layer::
995 include_view(app, MyView)
997 For small apps, it can also be used as a decorator::
999 @include_view(app)
1000 class MyView(AsyncRestView):
1001 ...
1003 Registering a view on several parents mounts its routes on each; calling
1004 ``include_view`` again with a parent the view is already registered on is
1005 a no-op.
1006 """
1007 if view_cls is not None:
1008 _init_view_cls_and_add_to_router(view_cls, parent_router)
1009 return view_cls
1011 def class_decorator(view_cls: V) -> V:
1012 _init_view_cls_and_add_to_router(view_cls, parent_router)
1013 return view_cls
1015 return class_decorator
1018def route(path: str, **api_route_kwargs: Any) -> Callable[..., Any]:
1019 """Decorator to mark a View method as an endpoint.
1020 The path and api_route_kwargs are passed into APIRouter.add_api_route(), see for example:
1021 https://fastapi.tiangolo.com/reference/apirouter/#fastapi.APIRouter.get
1023 Endpoints methods are later added as routes to the FastAPI app using `include_view()`
1024 """
1026 def store_args_decorator(func: Callable[..., Any]) -> Callable[..., Any]:
1027 # A new declaration takes precedence over metadata copied by wraps().
1028 func.__dict__.pop("_fr_typed_id_route", None)
1029 # Create a new attribute: '_api_route_args'
1030 func._api_route_args = (path, api_route_kwargs) # type: ignore[attr-defined]
1031 return func
1033 return store_args_decorator
1036def _typed_id_route(endpoint: Callable[..., Any]) -> Callable[..., Any]:
1037 """Mark a built-in item route whose copied path follows the view's ID type."""
1038 endpoint._fr_typed_id_route = True # type: ignore[attr-defined]
1039 return endpoint
1042def get(path: str, **api_route_kwargs: Any) -> Callable[..., Any]:
1043 """Decorator to mark a View method as a GET endpoint.
1045 Equivalent to::
1047 @route(path, methods=["GET"], status_code=200, ... )
1048 """
1049 api_route_kwargs.setdefault("methods", ["GET"])
1050 api_route_kwargs.setdefault("status_code", 200)
1051 return route(path, **api_route_kwargs)
1054def post(path: str, **api_route_kwargs: Any) -> Callable[..., Any]:
1055 """Decorator to mark a View method as a POST endpoint.
1057 Equivalent to::
1059 @route(path, methods=["POST"], status_code=201, ... )
1060 """
1061 api_route_kwargs.setdefault("methods", ["POST"])
1062 api_route_kwargs.setdefault("status_code", 201)
1063 return route(path, **api_route_kwargs)
1066def put(path: str, **api_route_kwargs: Any) -> Callable[..., Any]:
1067 """Decorator to mark a View method as a PUT endpoint.
1069 Equivalent to::
1071 @route(path, methods=["PUT"], status_code=200, ... )
1072 """
1073 api_route_kwargs.setdefault("methods", ["PUT"])
1074 api_route_kwargs.setdefault("status_code", 200)
1075 return route(path, **api_route_kwargs)
1078def patch(path: str, **api_route_kwargs: Any) -> Callable[..., Any]:
1079 """Decorator to mark a View method as a PATCH endpoint.
1081 Equivalent to::
1083 @route(path, methods=["PATCH"], status_code=200, ... )
1084 """
1085 api_route_kwargs.setdefault("methods", ["PATCH"])
1086 api_route_kwargs.setdefault("status_code", 200)
1087 return route(path, **api_route_kwargs)
1090def delete(path: str, **api_route_kwargs: Any) -> Callable[..., Any]:
1091 """Decorator to mark a View method as a DELETE endpoint.
1093 Equivalent to::
1095 @route(path, methods=["DELETE"], status_code=204, ... )
1096 """
1097 api_route_kwargs.setdefault("methods", ["DELETE"])
1098 api_route_kwargs.setdefault("status_code", 204)
1099 return route(path, **api_route_kwargs)
1102# Callable from a verb override, never overridden. The framework's own view
1103# classes define them; anything else in the MRO that does is a mistake.
1104_FINAL_DOMAIN_UTILITIES = ("make_new_object", "update_object", "save_object")
1106# Callable from a custom route, never overridden. Every generated CRUD route
1107# then runs ``authorize`` and, on writes, the commit bracket.
1108_FINAL_HANDLERS = (
1109 "handle_get_many",
1110 "handle_get_one",
1111 "handle_create",
1112 "handle_update",
1113 "handle_delete",
1114)
1115_WRITE_VERBS = ("create", "update", "delete")
1118def _final_handler_message(view_name: str, name: str, origin: str) -> str:
1119 """What a subclass that defines ``name`` is told to use instead."""
1120 verb = name.removeprefix("handle_")
1121 message = (
1122 f"{view_name} defines {name}{origin}, which is a final handler, not "
1123 f"a seam: it is the tier a custom route calls. Put domain logic in "
1124 f"the {verb} business method, a gate in authorize, "
1125 )
1126 if verb in _WRITE_VERBS:
1127 message += "a side effect in before_action_commit or after_action_commit, "
1128 message += f"and the HTTP contract in {verb}_endpoint."
1129 if verb in _WRITE_VERBS:
1130 message += (
1131 f" To put several writes in one commit, call {name} inside "
1132 "shared_write_action_commit()."
1133 )
1134 return message
1137# Replaced by ``pagination``. A view that still sets one is told what to write.
1138_REPLACED_PAGINATION_SETTINGS = ("paginated", "default_page_size", "max_page_size")
1141def _replaced_pagination_message(
1142 view_name: str, found: dict[str, tuple[Any, str]]
1143) -> str:
1144 """What a view that sets old pagination settings writes instead: one
1145 suggestion that carries every value it set. ``found`` maps each setting
1146 to its value and where it came from."""
1147 values = {name: value for name, (value, _) in found.items()}
1148 sizes = {
1149 name: values[name]
1150 for name in ("default_page_size", "max_page_size")
1151 if name in values
1152 }
1153 maximum = sizes.get("max_page_size")
1154 if (
1155 "default_page_size" not in sizes
1156 and isinstance(maximum, int)
1157 and maximum < DEFAULT_PAGE_SIZE
1158 ):
1159 # the default page size of 50 would be above the new maximum
1160 sizes = {"default_page_size": maximum, **sizes}
1161 if values.get("paginated") is False:
1162 instead = "Write 'pagination = None' to return every row."
1163 elif sizes:
1164 arguments = ", ".join(f"{name}={value!r}" for name, value in sizes.items())
1165 instead = (
1166 f"Write 'pagination = fr.NumberedPagination({arguments})', or call "
1167 f"'.replace({arguments})' on pagination settings that views share."
1168 )
1169 if "paginated" in values: 1169 ↛ 1170line 1169 didn't jump to line 1170 because the condition on line 1169 was never true
1170 instead += " Remove 'paginated': views paginate by default."
1171 else:
1172 instead = "Remove it: views paginate by default."
1173 names = ", ".join(f"{name}{origin}" for name, (_, origin) in found.items())
1174 verb = "was" if len(found) == 1 else "were"
1175 return (
1176 f"{view_name} sets {names}, which {verb} replaced by the 'pagination' "
1177 f"setting. {instead}"
1178 )
1181def _check_pagination_setting(cls: type) -> None:
1182 """Reject the replaced pagination settings, and a ``pagination`` that is
1183 not a settings instance. Runs when the class is defined and again at
1184 registration, so a value assigned to the class later is caught too."""
1185 # the old settings are no longer read; a definition would be silently
1186 # dead, and a dead max_page_size is a cap that is gone
1187 found: dict[str, tuple[Any, str]] = {}
1188 for klass in cls.__mro__:
1189 for name in _REPLACED_PAGINATION_SETTINGS:
1190 if name in vars(klass) and name not in found:
1191 origin = "" if klass is cls else f" (from {klass.__name__})"
1192 found[name] = (vars(klass)[name], origin)
1193 if found:
1194 raise RestlyConfigurationError(
1195 _replaced_pagination_message(cls.__name__, found)
1196 )
1197 pagination = getattr(cls, "pagination", None)
1198 if not (
1199 pagination is None or isinstance(pagination, (NumberedPagination, NoPagination))
1200 ):
1201 hint = (
1202 f" Create an instance: fr.{pagination.__name__}()."
1203 if pagination in (NumberedPagination, NoPagination)
1204 else ""
1205 )
1206 raise RestlyConfigurationError(
1207 f"{cls.__name__}.pagination must be a fr.NumberedPagination or "
1208 f"fr.NoPagination instance, or None, got {pagination!r}.{hint}"
1209 )
1212# Renamed before 1.0. Restly no longer reads the old names, so a definition
1213# would be silently dead, or would fail only on a request.
1214_RENAMED_SETTINGS = {"listing_param_schema": "schema_list_params"}
1215_RENAMED_METHODS = {
1216 "to_listing_response": "to_list_response",
1217 "to_react_admin_listing_response": "to_list_response",
1218 "to_response_schema": "to_single_response",
1219 "apply_query_params": "apply_list_params",
1220}
1221_RENAMED_METHOD_HINTS = {
1222 "to_listing_response": (
1223 " It takes only the list result: the list params are in "
1224 "list_result.list_params."
1225 ),
1226 "to_react_admin_listing_response": (
1227 " The list params are in list_result.list_params."
1228 ),
1229 "apply_query_params": " Its arguments are (query, list_params).",
1230}
1233def _route_parameters(route: Callable[..., Any]) -> Mapping[str, inspect.Parameter]:
1234 """The parameters of ``route``, without evaluating its annotations.
1236 On 3.14 a route may name a class that its module defines further down,
1237 because annotations are read only when asked for. Evaluating them while
1238 the view class is defined would raise ``NameError`` for that name.
1239 """
1240 if sys.version_info >= (3, 14): 1240 ↛ 1241line 1240 didn't jump to line 1241 because the condition on line 1240 was never true
1241 return inspect.signature(
1242 route, annotation_format=annotationlib.Format.FORWARDREF
1243 ).parameters
1244 return inspect.signature(route).parameters
1247def _check_renamed_names(cls: type) -> None:
1248 """Reject a setting, a method or a route parameter that still uses a name
1249 from before the rename. Runs when the class is defined and again at
1250 registration, so a value assigned to the class later is caught too."""
1251 for klass in cls.__mro__:
1252 origin = "" if klass is cls else f" (from {klass.__name__})"
1253 for old, new in _RENAMED_SETTINGS.items():
1254 if old in vars(klass):
1255 raise RestlyConfigurationError(
1256 f"{cls.__name__} sets {old}{origin}, which was renamed to "
1257 f"{new}. Rename the setting."
1258 )
1259 for old, new in _RENAMED_METHODS.items():
1260 if old in vars(klass):
1261 raise RestlyConfigurationError(
1262 f"{cls.__name__} defines {old}{origin}, which was renamed "
1263 f"to {new}.{_RENAMED_METHOD_HINTS.get(old, '')} Rename the "
1264 "method."
1265 )
1266 for name, value in vars(klass).items():
1267 if not hasattr(value, "_api_route_args"):
1268 continue
1269 if "query_params" in _route_parameters(value):
1270 raise RestlyConfigurationError(
1271 f"{cls.__name__}.{name}{origin} declares a query_params "
1272 "parameter, which was renamed to list_params. Rename the "
1273 "parameter, so the route still takes the filter, sort and "
1274 "page parameters."
1275 )
1278class BaseRestView(
1279 View, Generic[ModelT, ResponseSchemaT, CreateSchemaT, UpdateSchemaT, IdT]
1280):
1281 """
1282 Base class for RestView implementations.
1284 This class contains the common functionality shared between AsyncRestView
1285 and RestView, including schema definitions, model configuration, and
1286 common CRUD operation logic.
1287 """
1289 responses: ClassVar[dict[int | str, dict[str, Any]]] = {
1290 404: {"description": "Not found"}
1291 }
1293 schema: ClassVar[type[pydantic.BaseModel]]
1294 #: The class of everything that goes out: the single responses, the
1295 #: items of the list response, and the fields the list params filter and
1296 #: sort on. The relationships it names are the ones the view loads. When
1297 #: a view sets none, Restly builds it with
1298 #: :func:`~fastapi_restly.schemas.derive_schema_response`: the view's
1299 #: schema without its ``WriteOnly`` fields, named ``<Resource>Response``.
1300 #: A class that a view sets is used as it is.
1301 schema_response: ClassVar[type[pydantic.BaseModel]]
1302 # If 'schema_create' is not defined it will be created from 'schema'
1303 # using `create_model_without_read_only_fields()`.
1304 schema_create: ClassVar[type[pydantic.BaseModel]]
1305 schema_update: ClassVar[type[pydantic.BaseModel]]
1306 model: ClassVar[type[Any]]
1307 #: The clause every read on this view applies: list, count, and retrieve
1308 #: (a row outside it is 404). ``None`` (the default) falls back to the
1309 #: model's declared ``default_scope``; ``fr.clauses.UNSCOPED`` reads
1310 #: unscoped despite that default. Declaring a scope replaces the default,
1311 #: it does not stack on it; compose the replacement from the same leaves
1312 #: (``ItemClauses.trashed`` containing the tenant clause ``visible``
1313 #: contains). A rule that must hold under every scope is a session-level
1314 #: ``with_loader_criteria``, not a view concern. See the Scopes guide.
1315 scope: ClassVar[WhereClause | Unscoped | None] = None
1316 #: The type of the ``{id}`` path parameter on the default routes. ``None``
1317 #: (the default) takes the Python type of the model's primary key, and a
1318 #: composite key gets ``int``. Set a type to override it. Built-in item
1319 #: routes use Starlette's ``int`` or ``uuid`` path converter for those
1320 #: types. Integer paths accept non-negative digits. An unmatched value
1321 #: falls through to another route or returns 404, instead of 422.
1322 id_type: ClassVar[type[Any] | None] = None
1323 exclude_routes: ClassVar[Iterable[str | ViewRoute]] = ()
1324 #: Extra query-parameter keys to allow on list routes beyond those
1325 #: derived from the response class. Use this when a view reads a custom
1326 #: parameter from ``self.request`` (e.g. ``?verbose=true``). Without this,
1327 #: the strict unknown-key guard rejects the request with 422. A key that the
1328 #: route declares, on the endpoint method or in a dependency, or an
1329 #: ``APIKeyQuery`` key, needs no entry.
1330 extra_query_params: ClassVar[Iterable[str]] = ()
1331 #: How list endpoints paginate. The default :class:`NumberedPagination`
1332 #: takes ``page`` / ``page_size`` query parameters, runs the count query,
1333 #: and wraps the list in its ``envelope`` (:class:`PaginatedEnvelope`:
1334 #: ``data`` plus ``total_count`` / ``page`` / ``page_size`` /
1335 #: ``total_pages``). A :class:`NoPagination` returns every matching row
1336 #: with no count, in its ``envelope``; ``None`` is short for
1337 #: ``NoPagination()``, a plain :class:`Envelope` (``data`` only). A view
1338 #: inherits it, so a project base view sets it once; a view that differs
1339 #: changes one setting with ``replace()`` on the shared settings. For a
1340 #: shape no envelope model can express, such as a header, replace
1341 #: ``get_many_endpoint`` with a matching ``response_model`` (see
1342 #: :class:`AsyncReactAdminView`).
1343 pagination: ClassVar[NumberedPagination | NoPagination | None] = _DEFAULT_PAGINATION
1344 #: The list params (filter, sort, page) as a pydantic model, generated
1345 #: from ``schema_response`` and ``model`` by
1346 #: :func:`~fastapi_restly.query.derive_schema_list_params`. Any route
1347 #: method on the view that declares a ``list_params`` parameter takes it:
1348 #: typed for FastAPI and OpenAPI, and guarded against unknown keys like
1349 #: ``GET /``, so a custom list route (a trash route naming its own scope)
1350 #: reads the same list params. The route can take other query parameters
1351 #: beside it.
1352 schema_list_params: ClassVar[type[pydantic.BaseModel]]
1354 request: fastapi.Request
1356 def __init_subclass__(cls, **kwargs: Any) -> None:
1357 super().__init_subclass__(**kwargs)
1358 # validate at class definition, so a raw expression fails at
1359 # import instead of on the first read
1360 scope = cls.__dict__.get("scope")
1361 if not (scope is None or scope is UNSCOPED or isinstance(scope, WhereClause)):
1362 raise RestlyConfigurationError(
1363 f"{cls.__name__}.scope must be a WhereClause, "
1364 f"fr.clauses.UNSCOPED, or None, got {type(scope).__name__}; "
1365 "wrap a raw expression with where_clause()"
1366 )
1367 # the annotation accepts any class, so the mapping is checked here
1368 model = cls.__dict__.get("model")
1369 if model is not None and not is_mapped_class(model):
1370 label = getattr(model, "__name__", repr(model))
1371 raise RestlyConfigurationError(
1372 f"{cls.__name__}.model must be a SQLAlchemy mapped class, "
1373 f"got {label}, which has no mapper"
1374 )
1375 # build_query is removed; a definition would be silently dead code,
1376 # and dead visibility filtering is a security hole
1377 for klass in cls.__mro__:
1378 if "build_query" in vars(klass):
1379 origin = "" if klass is cls else f" (from {klass.__name__})"
1380 raise RestlyConfigurationError(
1381 f"{cls.__name__} defines build_query{origin}, which is "
1382 "removed and no longer called. Declare visibility as a "
1383 "clause: default_scope on the model's ClauseNamespace, or "
1384 "the scope attribute on the view. Reshape a list in "
1385 "apply_list_params, and relationship loading in "
1386 "get_relationship_loader_options. See Migrating from "
1387 "build_query in the Scopes guide."
1388 )
1389 # delete_object is removed; a stale soft-delete override would be
1390 # silently dead code, and a dead soft delete is a hard delete
1391 for klass in cls.__mro__:
1392 if "delete_object" in vars(klass):
1393 origin = "" if klass is cls else f" (from {klass.__name__})"
1394 raise RestlyConfigurationError(
1395 f"{cls.__name__} defines delete_object{origin}, which is "
1396 "removed and no longer called. Override delete instead: a "
1397 "soft delete flips a timestamp there. A raw row delete is "
1398 "fr.objects.delete_object / async_delete_object on "
1399 "self.session."
1400 )
1401 # the domain utilities are final: a stamp hooked there only covered
1402 # the verbs, and a definition here would shadow the framework's
1403 for klass in cls.__mro__:
1404 if klass.__module__.startswith("fastapi_restly."):
1405 continue
1406 for name in _FINAL_DOMAIN_UTILITIES:
1407 if name in vars(klass):
1408 origin = "" if klass is cls else f" (from {klass.__name__})"
1409 raise RestlyConfigurationError(
1410 f"{cls.__name__} defines {name}{origin}, which is a "
1411 "final domain utility, not a seam. Stamp a "
1412 "server-controlled field with a column default on the "
1413 "model (it then covers every write path); put "
1414 "payload-derived logic in the create / update verb; "
1415 "put a per-write side effect in before_action_commit."
1416 )
1417 # the handlers are final: an override re-implements the load,
1418 # authorize and commit order to reach something another seam gives
1419 # directly, and a generated route could then skip either
1420 for klass in cls.__mro__:
1421 if klass.__module__.startswith("fastapi_restly."):
1422 continue
1423 for name in _FINAL_HANDLERS:
1424 if name in vars(klass):
1425 origin = "" if klass is cls else f" (from {klass.__name__})"
1426 raise RestlyConfigurationError(
1427 _final_handler_message(cls.__name__, name, origin)
1428 )
1429 _check_pagination_setting(cls)
1430 _check_renamed_names(cls)
1432 def _apply_scope(self, query: Select[Any], scope: ReadScope) -> Select[Any]:
1433 # the one path every read takes, so retrieve, list and count cannot
1434 # disagree about which rows exist; not an override point
1435 if scope is None:
1436 scope = resolve_scope(self)
1437 else:
1438 # a per-read scope replaces the resolution rather than joining it
1439 scope = _checked_scope(scope, type(self).__name__, "a per-read scope")
1440 return apply_clauses(query, scope)
1442 def _narrowed_scope(self, scope: ReadScope, where: _ReadWhere) -> ReadScope:
1443 # the list handler is final, so it can fold where= into the scope it
1444 # forwards: get_many keeps one visibility input, and an override
1445 # keeps its signature
1446 if where is None:
1447 return scope
1448 label = type(self).__name__
1449 if scope is None:
1450 scope = resolve_scope(self)
1451 else:
1452 scope = _checked_scope(scope, label, "a per-read scope")
1453 return all_of(scope, _checked_where(where, label))
1455 def get_relationship_loader_options(self) -> list[Any]:
1456 """Loader options for the relationships the response class names.
1458 Returns recursive ``selectinload(...)`` options derived from
1459 :attr:`schema_response`, applied on reads (``get_one`` /
1460 ``get_many``) and on the post-write reload in ``save_object``.
1461 Override to eager-load relationships the response class does not
1462 name on both paths; append to
1463 ``super().get_relationship_loader_options()`` to keep the derived
1464 loads. See the "Relationship Loading and Async" how-to.
1465 """
1466 return _build_relationship_loader_options(
1467 self.model, _schema_response(type(self))
1468 )
1470 def _get_response_reload_statement(self, obj: Any) -> Any:
1471 """The SELECT that makes ``obj`` serializable, or ``None`` if it already is.
1473 Called by ``save_object`` *after* the flush and refresh, because the
1474 refresh is itself what leaves relationships unloaded; asking earlier
1475 would see a value the refresh is about to discard.
1476 """
1477 if _schema_relationships_are_loaded(
1478 obj, self.model, _schema_response(type(self))
1479 ):
1480 return None
1481 options = self.get_relationship_loader_options()
1482 if not options: 1482 ↛ 1483line 1482 didn't jump to line 1483 because the condition on line 1482 was never true
1483 return None
1484 return _relationship_reload_statement(obj, options)
1486 def _reject_unknown_query_params(self) -> None:
1487 """Reject any query-string key that neither the list params nor the
1488 route reads.
1490 Runs before every route that declares ``list_params`` (see
1491 :attr:`schema_list_params`).
1493 The list params dependency (:func:`_list_params_dependency`) reads
1494 their keys and ignores any other, which would let typoed
1495 filters or unsupported operators (e.g.
1496 ``active__gte=true`` on a boolean column where the schema does not
1497 emit a range operator) widen the result set without telling the
1498 caller. We treat unknown keys as a validation error instead, mirroring
1499 FastAPI's 422 envelope shape so the response is consistent with
1500 bound-violation errors.
1502 No-op when there is no live request, as when code calls a route
1503 method directly, outside an HTTP request. Then there is no URL to
1504 check, and the caller is responsible for what it passes.
1505 """
1506 request = getattr(self, "request", None)
1507 if request is None: 1507 ↛ 1508line 1507 didn't jump to line 1508 because the condition on line 1507 was never true
1508 return
1509 schema_list_params = getattr(self, "schema_list_params", None)
1510 if schema_list_params is None: 1510 ↛ 1511line 1510 didn't jump to line 1511 because the condition on line 1510 was never true
1511 return
1512 list_keys = {
1513 _query_key(name, field)
1514 for name, field in schema_list_params.model_fields.items()
1515 }
1516 reject_unknown_query_keys(request, list_keys | set(self.extra_query_params))
1518 def to_single_response(self, obj: ModelT | ResponseSchemaT) -> ResponseSchemaT:
1519 """Serialize one ORM object to the view's :attr:`schema_response`.
1521 By default the response class is the view's schema without its
1522 WriteOnly fields, named like ``UserResponse``. An override may return
1523 an instance of another model, such as :attr:`schema`. Only the fields
1524 of the response class go out. Restly copies the values of an instance
1525 of a subclass of the response class, and of :attr:`schema` when
1526 Restly derived the response class, without validating them again. It
1527 validates any other model into the response class from its
1528 attributes.
1530 The view loads only the relationships that the response class names.
1531 An override that builds :attr:`schema` from the ORM object also reads
1532 the relationships of its WriteOnly fields. On an async view that fails
1533 with ``MissingGreenlet``: add them in
1534 :meth:`get_relationship_loader_options`, or build the response class
1535 instead.
1537 The ORM path below validates through the response class, so a view's
1538 schema that declares a WriteOnly field the ORM object doesn't carry
1539 (e.g. ``password`` backed by a ``password_hash`` column) doesn't fail
1540 response validation.
1541 """
1542 schema_response = _schema_response(type(self))
1543 response = _as_response(schema_response, obj)
1544 if isinstance(response, schema_response):
1545 return cast(ResponseSchemaT, response)
1547 # Build a payload of raw attribute values keyed by schema field name;
1548 # re-validating it below serializes each field through its own type. The
1549 # response schema's ``from_attributes`` config resolves nested schemas and
1550 # reference types (IDRef/IDSchema) straight from the ORM rows, with no
1551 # view-layer special-casing. Alias rendering happens when FastAPI
1552 # serializes the response model.
1553 payload: dict[str, Any] = {}
1554 for field_name, field_info in schema_response.model_fields.items():
1555 if hasattr(obj, field_name):
1556 payload[field_name] = getattr(obj, field_name)
1557 elif field_info.alias and hasattr(obj, field_info.alias): 1557 ↛ 1558line 1557 didn't jump to line 1558 because the condition on line 1557 was never true
1558 payload[field_name] = getattr(obj, field_info.alias)
1560 adapter = _response_adapter(schema_response)
1561 return cast(
1562 ResponseSchemaT,
1563 adapter.validate_python(payload, by_alias=False, by_name=True),
1564 )
1566 def _single_response(self, obj: Any) -> pydantic.BaseModel:
1567 """One object as an instance of the response class. An override of
1568 :meth:`to_single_response` may return an instance of another model."""
1569 return _as_response(_schema_response(type(self)), self.to_single_response(obj))
1571 @staticmethod
1572 def _to_query_params(list_params: Any) -> QueryParams:
1573 if isinstance(list_params, QueryParams):
1574 return list_params
1575 if isinstance(list_params, pydantic.BaseModel):
1576 dumped = list_params.model_dump(
1577 exclude_none=True, by_alias=True, mode="json"
1578 )
1579 return QueryParams({k: str(v) for k, v in dumped.items()})
1580 if isinstance(list_params, dict): 1580 ↛ 1582line 1580 didn't jump to line 1582 because the condition on line 1580 was always true
1581 return QueryParams({k: str(v) for k, v in list_params.items()})
1582 return QueryParams(list_params)
1584 def to_list_response(self, list_result: ListResult[ModelT]) -> Any:
1585 """Build the list response body: an instance of the envelope model.
1587 The view fills its :attr:`pagination`'s ``envelope``: by default a
1588 :class:`PaginatedEnvelope` (``data`` plus ``total_count`` / ``page`` /
1589 ``page_size`` / ``total_pages``), or an :class:`Envelope` (``data``
1590 only) for a view without pagination. The route's ``response_model`` is
1591 the same envelope, fixed from :attr:`pagination` at registration. To
1592 change the shape, set the pagination's ``envelope``. For a
1593 ``Content-Range`` header, replace ``get_many_endpoint`` with a matching
1594 ``response_model`` (as :class:`AsyncReactAdminView` does), not by
1595 overriding this method alone, which would fail response validation.
1597 The page and page size come from ``list_result.list_params``.
1598 """
1599 data = [self._single_response(obj) for obj in list_result.objects]
1600 pagination = self.pagination
1601 envelope = derive_schema_list_response(
1602 _schema_response(type(self)), pagination=pagination
1603 )
1604 if not isinstance(pagination, NumberedPagination):
1605 return _build_envelope(envelope, _unpaginated_page_info(data))
1606 params = self._to_query_params(list_result.list_params)
1607 page_info = _numbered_page_info(
1608 total_count=list_result.total_count or 0,
1609 page=int(params.get(pagination.page_query_param) or 1),
1610 page_size=int(
1611 params.get(pagination.page_size_query_param)
1612 or pagination.default_page_size
1613 ),
1614 )
1615 return _build_envelope(envelope, {"data": data, **page_info})
1617 def to_response(
1618 self, result: Any, shape: ResponseShape = ResponseShape.SINGLE
1619 ) -> Any:
1620 """Endpoint-method response boundary.
1622 ``shape`` selects the wire form, and with it what ``result`` is: one
1623 object for ``SINGLE`` (:meth:`to_single_response`), a
1624 :class:`ListResult` for ``LIST`` (:meth:`to_list_response`), and
1625 nothing for ``EMPTY``. A plain Python list is not a valid ``result``.
1626 ``shape`` is not the write-action name. Override for envelopes or
1627 shape-wide status behavior; per-endpoint projections belong in the
1628 endpoint method.
1629 """
1630 if shape is ResponseShape.EMPTY:
1631 return fastapi.Response(status_code=204)
1632 if shape is ResponseShape.LIST:
1633 return self.to_list_response(result)
1634 return self._single_response(result)
1636 def snapshot(self, obj: Any) -> dict[str, Any]:
1637 """Frozen capture of an object's already-loaded column values, passed as
1638 ``old`` to ``before_action_commit`` / ``after_action_commit`` for dirty detection.
1639 Override to change what ``old`` captures (e.g. include a relationship's
1640 prior state); the default delegates to
1641 :func:`~fastapi_restly.objects.snapshot`.
1642 """
1643 return _object_snapshot(obj)
1645 @classmethod
1646 def _check_declared_schema_list_params(cls) -> None:
1647 """Reject declared list params that ``derive_schema_list_params`` made
1648 for other pagination settings: they read other page parameters than
1649 the view applies, or none. Hand-written list params are the author's
1650 to keep in step."""
1651 created_with = _created_with(cls.schema_list_params)
1652 if created_with is _UNKNOWN:
1653 return
1654 if _query_settings(created_with) != _query_settings(cls.pagination):
1655 raise RestlyConfigurationError(
1656 f"{cls.__name__}.schema_list_params was created for other "
1657 "pagination settings than the view's. Create it with "
1658 f"pagination={cls.__name__}.pagination, or leave it out to "
1659 "have it generated."
1660 )
1662 @classmethod
1663 def before_include_view(cls):
1664 """
1665 Apply type annotations needed for FastAPI, before creating an APIRouter from
1666 this view and registering it.
1668 This function can be overridden to further tweak the endpoints before they
1669 are added to FastAPI.
1670 """
1671 _check_pagination_setting(cls)
1672 _check_renamed_names(cls)
1674 # A declared schema is inherited like any attribute. One that Restly
1675 # generated for a parent is rebuilt, since the subclass may change
1676 # what it was derived from.
1677 if _needs_generating(cls, "schema"):
1678 if not hasattr(cls, "model"): 1678 ↛ 1679line 1678 didn't jump to line 1679 because the condition on line 1678 was never true
1679 raise ValueError(
1680 f"'{cls.__name__}.model' must be specified to auto-generate schema"
1681 )
1682 cls.schema = derive_schema(cls.model)
1683 _mark_generated(cls, "schema")
1685 if _needs_generating(cls, "schema_response"):
1686 cls.schema_response = derive_schema_response(cls.schema)
1687 _mark_generated(cls, "schema_response")
1689 if _needs_generating(cls, "schema_list_params"):
1690 if not hasattr(cls, "model"):
1691 raise ValueError(
1692 f"'{cls.__name__}.model' must be specified: it is needed to "
1693 "generate the list params."
1694 )
1695 cls.schema_list_params = derive_schema_list_params(
1696 cls.schema_response, cls.model, pagination=cls.pagination
1697 )
1698 _mark_generated(cls, "schema_list_params")
1699 else:
1700 cls._check_declared_schema_list_params()
1701 if _needs_generating(cls, "schema_create"):
1702 cls.schema_create = cast(
1703 type[CreateSchemaT],
1704 create_model_without_read_only_fields(
1705 cls.schema, omit=_generated_primary_key_fields(cls.model)
1706 ),
1707 )
1708 _mark_generated(cls, "schema_create")
1709 if _needs_generating(cls, "schema_update"):
1710 cls.schema_update = cast(
1711 type[UpdateSchemaT],
1712 create_model_with_optional_fields(
1713 cls.schema, omit=_generated_primary_key_fields(cls.model)
1714 ),
1715 )
1716 _mark_generated(cls, "schema_update")
1718 response_schema = cls.schema_response
1719 id_type = _view_id_type(cls)
1720 if id_type is int:
1721 item_path = "/{id:int}"
1722 elif id_type is UUID:
1723 item_path = "/{id:uuid}"
1724 else:
1725 item_path = "/{id}"
1726 for endpoint in cls.__dict__.values():
1727 if getattr(endpoint, "_fr_typed_id_route", False):
1728 _, route_kwargs = endpoint._api_route_args
1729 endpoint._api_route_args = (item_path, route_kwargs.copy())
1731 # Only annotate if the methods exist (they will be overridden in subclasses)
1732 list_response_annotation: Any = derive_schema_list_response(
1733 response_schema, pagination=cls.pagination
1734 )
1736 # Every route that declares ``list_params`` takes the view's list
1737 # params: ``GET /`` and any custom list route alike, guarded the same
1738 # way. The guard checks keys against ``schema_list_params`` whatever
1739 # the annotation says, so they replace an explicit annotation too. A
1740 # dependency reads them rather than a FastAPI query model, which
1741 # collapses next to any other query parameter (see
1742 # :func:`_list_params_dependency`).
1743 list_params = Annotated[
1744 cls.schema_list_params,
1745 fastapi.Depends(_list_params_dependency(cls.schema_list_params)),
1746 ]
1747 for name, route in list(cls.__dict__.items()):
1748 if not hasattr(route, "_api_route_args"):
1749 continue
1750 if "list_params" not in _route_parameters(route):
1751 continue
1752 if not getattr(route, "_fr_list_guard", False):
1753 route = _guard_list_params(route)
1754 setattr(cls, name, route)
1755 _annotate(route, overwrite=True, list_params=list_params)
1757 # The ``*_endpoint`` methods are defined on AsyncRestView/RestView
1758 # subclasses and may be excluded by ``exclude_routes``, so they aren't
1759 # visible on BaseRestView. ``getattr`` keeps pyright happy without
1760 # falsely advertising them on the base class.
1761 if (ep := getattr(cls, "get_many_endpoint", None)) is not None: 1761 ↛ 1763line 1761 didn't jump to line 1763 because the condition on line 1761 was always true
1762 _annotate(ep, return_annotation=list_response_annotation)
1763 if (ep := getattr(cls, "get_one_endpoint", None)) is not None: 1763 ↛ 1765line 1763 didn't jump to line 1765 because the condition on line 1763 was always true
1764 _annotate(ep, return_annotation=response_schema, id=id_type)
1765 if (ep := getattr(cls, "create_endpoint", None)) is not None: 1765 ↛ 1769line 1765 didn't jump to line 1769 because the condition on line 1765 was always true
1766 _annotate(
1767 ep, return_annotation=response_schema, schema_obj=cls.schema_create
1768 )
1769 if (ep := getattr(cls, "update_endpoint", None)) is not None: 1769 ↛ 1776line 1769 didn't jump to line 1776 because the condition on line 1769 was always true
1770 _annotate(
1771 ep,
1772 return_annotation=response_schema,
1773 schema_obj=cls.schema_update,
1774 id=id_type,
1775 )
1776 if (ep := getattr(cls, "delete_endpoint", None)) is not None: 1776 ↛ 1778line 1776 didn't jump to line 1778 because the condition on line 1776 was always true
1777 _annotate(ep, return_annotation=fastapi.Response, id=id_type)
1778 _exclude_routes(cls)
1781#: What registration builds each class of a view from.
1782_BUILT_FROM = {
1783 "schema": "model",
1784 "schema_response": "schema",
1785 "schema_list_params": "schema_response",
1786 "schema_create": "schema",
1787 "schema_update": "schema",
1788}
1791def _mark_generated(cls: type, name: str) -> None:
1792 generated = cls.__dict__.get("_fr_generated", frozenset())
1793 cls._fr_generated = generated | {name} # type: ignore[attr-defined]
1796def _declared_at(cls: type, name: str) -> int | None:
1797 """Position in the MRO of the class that declares ``name``, or None.
1799 A value that registration generated is not a declaration.
1800 """
1801 for index, klass in enumerate(cls.__mro__):
1802 if name in klass.__dict__ and name not in klass.__dict__.get(
1803 "_fr_generated", ()
1804 ):
1805 return index
1806 return None
1809def _decided_at(cls: type, name: str) -> int | None:
1810 """Position in the MRO of the class that decides the value of ``name``.
1812 A declared value is decided where it is declared, unless what it is built
1813 from is decided nearer to ``cls``: then registration rebuilds it. A
1814 generated value is decided where its source is decided. So a copy that
1815 registration wrote into a subclass does not count as a change.
1816 """
1817 declared = _declared_at(cls, name)
1818 source = _BUILT_FROM.get(name)
1819 if source is None:
1820 return declared
1821 source_at = _decided_at(cls, source)
1822 if declared is None:
1823 return source_at
1824 # A declared schema stays, whatever model a subclass sets.
1825 if name != "schema" and source_at is not None and source_at < declared:
1826 return source_at
1827 return declared
1830def _needs_generating(cls: type, name: str) -> bool:
1831 """Whether registration must build ``name`` for this view class.
1833 A declared value is inherited. Registration builds it when no class
1834 declares it, or when what it is built from is decided nearer to ``cls``,
1835 as in a subclass that declares a new ``schema``.
1836 """
1837 if name in cls.__dict__.get("_fr_generated", ()):
1838 return False
1839 declared = _declared_at(cls, name)
1840 return declared is None or _decided_at(cls, name) != declared
1843def _schema_response(view_cls: type) -> type[pydantic.BaseModel]:
1844 """The response class of a view.
1846 Registration sets ``schema_response``. A view that is not registered
1847 yet, or a subclass that changes ``schema``, gets the class that
1848 registration would set.
1849 """
1850 if _needs_generating(view_cls, "schema_response"):
1851 return derive_schema_response(view_cls.schema)
1852 return view_cls.schema_response
1855#: A ``BaseRestView`` at any parameterization, for the signature below.
1856_AnyRestView = BaseRestView[Any, Any, Any, Any, Any]
1859def resolve_scope(target: type[Any] | _AnyRestView) -> WhereClause | Unscoped:
1860 """The scope a read applies, resolved down the ladder.
1862 Pass a view, class or instance, for the visibility that view's reads
1863 apply: its declared
1864 :attr:`scope <fastapi_restly.views.BaseRestView.scope>`, the model's
1865 ``default_scope`` when the view declares none, and
1866 ``fr.clauses.UNSCOPED`` when neither does. Pass a mapped model class
1867 for the model rung alone, which is what every reference check applies.
1868 The two answers differ wherever a view declares a scope, so code off
1869 the request path that wants what the API shows passes the view.
1871 The result is a clause or ``fr.clauses.UNSCOPED``, never ``None``, so
1872 it composes without a branch::
1874 query = fr.apply_clauses(
1875 select(Task).where(Task.project_id == id),
1876 fr.resolve_scope(TaskView),
1877 )
1879 Every read resolves here, so a route that applies the result sees the
1880 same rows as
1881 :meth:`get_one <fastapi_restly.views.RestView.get_one>` and
1882 :meth:`get_many <fastapi_restly.views.RestView.get_many>`. It resolves
1883 the scope and does not apply it: applying stays with the framework,
1884 and there is no apply-side override point.
1886 :param target: a view class or instance, or a mapped model class.
1887 :raises RestlyConfigurationError: if the declared scope is not a clause.
1888 :raises TypeError: if ``target`` is neither a view nor a mapped model.
1889 """
1890 label = target.__name__ if isinstance(target, type) else type(target).__name__
1891 if is_mapped_class(target):
1892 return _model_scope(target)
1893 if not isinstance(target, BaseRestView) and not (
1894 isinstance(target, type) and issubclass(target, BaseRestView)
1895 ):
1896 raise TypeError(
1897 "resolve_scope() takes a RestView / AsyncRestView class or instance, "
1898 f"or a mapped model class, got {label}"
1899 )
1900 declared = target.scope
1901 if declared is None:
1902 return _model_scope(target.model)
1903 if declared is UNSCOPED:
1904 return UNSCOPED
1905 # backstop for a post-definition assignment; the declared form is
1906 # validated in __init_subclass__
1907 return _checked_scope(declared, label, "the scope")
1910def _model_scope(model: type[Any]) -> WhereClause | Unscoped:
1911 declared = _default_scope(model)
1912 return UNSCOPED if declared is None else declared
1915def _checked_scope(scope: Any, label: str, what: str) -> WhereClause | Unscoped:
1916 if scope is UNSCOPED or isinstance(scope, WhereClause):
1917 return scope
1918 raise RestlyConfigurationError(
1919 f"{label}: {what} must be a WhereClause or fr.clauses.UNSCOPED, got "
1920 f"{type(scope).__name__}; wrap a raw expression with where_clause()"
1921 )
1924def _checked_where(where: Any, label: str) -> WhereClause | Unscoped:
1925 if where is UNSCOPED or isinstance(where, WhereClause):
1926 return where
1927 if isinstance(where, ColumnElement):
1928 return where_clause(where)
1929 if type(where) is bool:
1930 # ``obj.slug == slug`` on a loaded object is a Python bool, and
1931 # SQLAlchemy renders it as WHERE true, which narrows nothing
1932 raise TypeError(
1933 f"{label}: where= got a bool, not a SQL expression. A comparison "
1934 "on an instance attribute is a bool; compare the column instead "
1935 "(Model.slug == slug)."
1936 )
1937 raise TypeError(
1938 f"{label}: where= must be a SQLAlchemy boolean expression or a clause, "
1939 f"got {type(where).__name__}"
1940 )
1943def _identity_criterion(model_cls: type[Any], id: Any) -> ColumnElement[bool]:
1944 """The predicate that picks the one row ``get_one`` loads.
1946 The primary key is the default: ``get_one(5)`` means ``pk == 5``. A
1947 SQLAlchemy boolean expression replaces that default, so a natural-key
1948 route loads through the same scope, loader options and 404, and a
1949 composite key is addressable at all. The criterion narrows inside the
1950 scope; only ``scope=`` replaces what the view can see.
1951 """
1952 if isinstance(id, ColumnElement):
1953 return id
1954 if isinstance(id, WhereClause):
1955 raise TypeError(
1956 f"{model_cls.__name__}: a clause is a scope, not a row identity. "
1957 "Call it for its expression (ItemClauses.published()), or "
1958 "pass it as scope= to replace the view scope for this read."
1959 )
1960 if type(id) is bool:
1961 # ``obj.slug == slug`` on a loaded object is a Python bool, and
1962 # SQLAlchemy renders it as WHERE true, which matches the whole scope
1963 raise TypeError(
1964 f"{model_cls.__name__}: got a bool, not an id or a SQL expression. "
1965 "A comparison on an instance attribute is a bool; compare the "
1966 "column instead (Model.slug == slug)."
1967 )
1968 pk_cols = sa_inspect(model_cls).primary_key
1969 if len(pk_cols) != 1:
1970 raise NotImplementedError(
1971 f"{model_cls.__name__} has a composite primary key, so an id does "
1972 "not name one row; pass a predicate instead, e.g. "
1973 "get_one(sqlalchemy.and_(Model.a == a, Model.b == b))."
1974 )
1975 return pk_cols[0] == id
1978def _not_found_message(model_cls: type[Any], id: Any) -> str:
1979 # a predicate is not echoed: the 404 body is no place for SQL
1980 if isinstance(id, ColumnElement):
1981 return f"{model_cls.__name__} was not found"
1982 return f"{model_cls.__name__} with id {id!r} was not found"
1985def reject_unknown_query_keys(request: fastapi.Request, allowed: set[str]) -> None:
1986 """Raise a 422 naming every query key that is not in ``allowed``.
1988 The dialect decides what is allowed (the list params for the default
1989 one, ``sort``/``range``/``filter`` for react-admin); the envelope is the
1990 same either way, and mirrors FastAPI's own validation shape. A key that
1991 the route reads elsewhere is never unknown (see :func:`_route_query_keys`).
1992 """
1993 unknown = set(request.query_params.keys()) - allowed
1994 if unknown:
1995 unknown -= _route_query_keys(request)
1996 if not unknown:
1997 return
1998 detail = [
1999 {
2000 "type": "extra_forbidden",
2001 "loc": ["query", key],
2002 "msg": f"Unknown query parameter {key!r}",
2003 "input": request.query_params.get(key),
2004 }
2005 for key in sorted(unknown)
2006 ]
2007 raise fastapi.HTTPException(status_code=422, detail=detail)
2010def _route_query_keys(request: fastapi.Request) -> set[str]:
2011 """The query keys that the matched route reads.
2013 Every query parameter its dependency tree declares, at any level: the
2014 app or a router, the view's ``dependencies``, a class attribute, the
2015 endpoint. Also the key of a query API key scheme such as
2016 ``APIKeyQuery``, which reads the query string without declaring a
2017 parameter. The guard walks the tree only for a key that its dialect
2018 does not allow.
2019 """
2020 dependant = getattr(matched_route(request.scope), "dependant", None)
2021 pending = [dependant] if dependant is not None else []
2022 keys: set[str] = set()
2023 while pending:
2024 current = pending.pop()
2025 for field in current.query_params:
2026 # the name FastAPI reads the value from
2027 keys.add(getattr(field, "validation_alias", None) or field.alias)
2028 call = current.call
2029 if isinstance(call, APIKeyBase) and call.model.in_ == APIKeyIn.query:
2030 keys.add(call.model.name)
2031 pending.extend(current.dependencies)
2032 return keys
2035def _list_params_dependency(
2036 params_model: type[pydantic.BaseModel],
2037) -> Callable[..., pydantic.BaseModel]:
2038 """A dependency that validates ``params_model`` from the query string.
2040 FastAPI splits a query model into its fields only when it is the sole
2041 query parameter: at runtime per function, in OpenAPI across the route.
2042 Any other query parameter therefore collapses it into one required
2043 ``list_params`` value. This reads the fields itself, the way FastAPI
2044 hands a query model its input: every value of a repeated key for a
2045 sequence field, the last value otherwise, a default that is not None,
2046 and any other key as sent. :func:`_list_openapi_parameters`
2047 documents the fields.
2048 """
2049 fields = [
2050 (_query_key(name, field), field, _reads_every_value(field))
2051 for name, field in params_model.model_fields.items()
2052 ]
2053 field_keys = {key for key, _, _ in fields}
2055 def list_params(request: fastapi.Request, first: Any = None) -> pydantic.BaseModel:
2056 query = request.query_params
2057 data: dict[str, Any] = {}
2058 for key, field, every_value in fields:
2059 if key in query:
2060 data[key] = query.getlist(key) if every_value else query[key]
2061 elif not field.is_required(): 2061 ↛ 2058line 2061 didn't jump to line 2058 because the condition on line 2061 was always true
2062 default = field.get_default()
2063 if default is not None:
2064 data[key] = default
2065 for key in query.keys():
2066 if key not in field_keys:
2067 values = query.getlist(key)
2068 data[key] = values[0] if len(values) == 1 else values
2069 try:
2070 return params_model.model_validate(data)
2071 except pydantic.ValidationError as exc:
2072 errors = [
2073 {**error, "loc": ("query", *error["loc"])}
2074 for error in exc.errors(include_url=False)
2075 ]
2076 raise RequestValidationError(errors) from None
2078 parameters = [
2079 inspect.Parameter(
2080 "request", inspect.Parameter.KEYWORD_ONLY, annotation=fastapi.Request
2081 )
2082 ]
2083 if fields:
2084 # FastAPI documents a route's 422 only when it reads a parameter of
2085 # the route itself. It reads the first key here, undocumented and
2086 # unchecked; the model validates the value with the rest.
2087 marker = fastapi.Query(alias=fields[0][0], include_in_schema=False)
2088 parameters.append(
2089 inspect.Parameter(
2090 "first",
2091 inspect.Parameter.KEYWORD_ONLY,
2092 default=None,
2093 annotation=Annotated[Any, marker],
2094 )
2095 )
2096 list_params.__signature__ = inspect.Signature(parameters) # type: ignore[attr-defined]
2097 return list_params
2100def _query_key(name: str, field: FieldInfo) -> str:
2101 """The query key of a list params field: its validation alias, its alias, or
2102 its name, as FastAPI reads a query model."""
2103 if isinstance(field.validation_alias, str):
2104 return field.validation_alias
2105 return field.alias or name
2108def _reads_every_value(field: FieldInfo) -> bool:
2109 """Whether a list params field reads every value of a repeated query key.
2111 FastAPI's rule for a query model: a sequence or set type, alone or in a
2112 union, unless the value is JSON.
2113 """
2114 if any(isinstance(item, pydantic.Json) for item in field.metadata): 2114 ↛ 2115line 2114 didn't jump to line 2115 because the condition on line 2114 was never true
2115 return False
2116 return _is_sequence_annotation(field.annotation)
2119def _is_sequence_annotation(annotation: Any) -> bool:
2120 if get_origin(annotation) in (Union, types.UnionType):
2121 return any(_is_sequence_annotation(arg) for arg in get_args(annotation))
2122 return any(
2123 isinstance(candidate, type)
2124 and get_origin(candidate) is None
2125 and issubclass(candidate, (Sequence, Set))
2126 and not issubclass(candidate, (str, bytes))
2127 for candidate in (annotation, get_origin(annotation))
2128 )
2131def _list_openapi_parameters(
2132 params_model: type[pydantic.BaseModel],
2133) -> list[dict[str, Any]]:
2134 """One OpenAPI query parameter per field of ``params_model``.
2136 The entries FastAPI writes for a query model it splits into fields, so
2137 the list route documents the same whatever else the route reads. Built from
2138 the model's JSON schema, with references inlined: the document's
2139 components do not hold the model's definitions.
2140 """
2141 schema = params_model.model_json_schema()
2142 definitions = schema.get("$defs", {})
2143 properties = schema.get("properties", {})
2144 required = set(schema.get("required", ()))
2145 parameters: list[dict[str, Any]] = []
2146 for name, field in params_model.model_fields.items():
2147 key = _query_key(name, field)
2148 field_schema = _inline_definitions(
2149 properties.get(key, properties.get(name, {})), definitions
2150 )
2151 field_schema["title"] = field.title or key.title().replace("_", " ")
2152 parameter: dict[str, Any] = {
2153 "name": key,
2154 "in": "query",
2155 "required": key in required,
2156 "schema": field_schema,
2157 }
2158 if field.description:
2159 parameter["description"] = field.description
2160 if field.deprecated: 2160 ↛ 2161line 2160 didn't jump to line 2161 because the condition on line 2160 was never true
2161 parameter["deprecated"] = True
2162 parameters.append(parameter)
2163 return parameters
2166def _inline_definitions(schema: Any, definitions: dict[str, Any]) -> Any:
2167 """``schema`` with each ``#/$defs/...`` reference replaced by its target."""
2168 if isinstance(schema, list):
2169 return [_inline_definitions(item, definitions) for item in schema]
2170 if not isinstance(schema, dict):
2171 return schema
2172 inlined = {
2173 key: _inline_definitions(value, definitions)
2174 for key, value in schema.items()
2175 if key != "$ref"
2176 }
2177 reference = schema.get("$ref")
2178 if isinstance(reference, str) and reference.startswith("#/$defs/"): 2178 ↛ 2179line 2178 didn't jump to line 2179 because the condition on line 2178 was never true
2179 target = definitions[reference.removeprefix("#/$defs/")]
2180 return {**_inline_definitions(target, definitions), **inlined}
2181 return inlined
2184def _guard_list_params(route: Callable) -> Callable:
2185 """Run the unknown-key guard before a route that takes ``list_params``.
2187 A copy, like the parent-endpoint copies: ``functools.wraps`` carries the
2188 route args and the signature, and the marker keeps a subclass's copy
2189 from being wrapped twice.
2190 """
2191 if inspect.iscoroutinefunction(route):
2193 @functools.wraps(route)
2194 async def _async_guarded(self, *args, **kwargs):
2195 self._reject_unknown_query_params()
2196 return await route(self, *args, **kwargs)
2198 guarded: Callable = _async_guarded
2199 else:
2201 @functools.wraps(route)
2202 def _sync_guarded(self, *args, **kwargs):
2203 self._reject_unknown_query_params()
2204 return route(self, *args, **kwargs)
2206 guarded = _sync_guarded
2208 guarded._fr_list_guard = True # type: ignore[attr-defined]
2209 return guarded
2212def _exclude_routes(cls: type[BaseRestView[Any, Any, Any, Any, Any]]):
2213 for route_name in cls.exclude_routes:
2214 method_name = (
2215 route_name.value if isinstance(route_name, ViewRoute) else route_name
2216 )
2217 # @route decorator adds `_api_route_args` to a method to create the route later.
2218 # By removing it from the method, the method will no longer be added as a route.
2219 view_func = getattr(cls, method_name, None)
2220 if view_func is not None and hasattr(view_func, "_api_route_args"):
2221 del view_func._api_route_args
2222 continue
2223 # Not a live route on this class. Tolerate an exclusion that is *already*
2224 # satisfied: a subclass that inherits ``exclude_routes`` from a parent
2225 # which already excluded the route never receives a routable copy, so
2226 # there is nothing to strip. The name is still a genuine route elsewhere
2227 # in the lineage; only raise when it is no route at all (a typo, or the
2228 # business method name instead of the ``*_endpoint`` route name).
2229 if not _is_route_name_in_lineage(cls, method_name):
2230 raise AttributeError(f"{method_name!r} is not a route on {cls.__name__}")
2233def _is_route_name_in_lineage(
2234 cls: type[BaseRestView[Any, Any, Any, Any, Any]], method_name: str
2235) -> bool:
2236 """True if any class in ``cls``'s MRO defines a routable endpoint of this
2237 name, so the name is a real route that may merely be already-excluded here.
2238 """
2239 return any(
2240 hasattr(klass.__dict__.get(method_name), "_api_route_args")
2241 for klass in cls.mro()
2242 )
2245def _init_view_cls_and_add_to_router(
2246 view_cls: type[View], parent_router: fastapi.APIRouter | fastapi.FastAPI
2247):
2248 """
2249 FastAPI builds routes from callable signatures and annotations. View classes
2250 need a small adapter layer before registration; that preparation happens here.
2252 FastAPI does a lot with annotations. For example, accepted or returned JSON is
2253 often described with Pydantic classes like this:
2255 def my_endpoint(foo: FooSchema) -> FooSchema:
2257 Most of this preparation sets the correct annotations on inherited class
2258 methods.
2260 The class-level preparation (copying parent endpoints, renaming, annotating,
2261 schema generation, dataclass-style __init__) only runs once per View class —
2262 subsequent calls to ``include_view()`` reuse the prepared class and only
2263 construct a fresh APIRouter to mount on the new parent. This makes
2264 registering the same view on *different* routers safe (e.g. a public app and
2265 an admin app, or ``/v1`` and ``/v2`` sub-apps).
2267 Re-registering the same view on the *same* parent is a no-op (flagged by
2268 the opt-in misuse lint): each parent tracks the view classes already
2269 mounted on it, so a duplicate ``include_view()`` call (double import,
2270 decorator plus explicit call) cannot mount the same routes twice.
2271 """
2272 # A FastAPI app delegates include_router to its .router; track mounted
2273 # views on the router so app and app.router count as one parent.
2274 holder = (
2275 parent_router.router
2276 if isinstance(parent_router, fastapi.FastAPI)
2277 else parent_router
2278 )
2279 mounted: set[type[View]] = getattr(holder, "_fr_mounted_views", None) or set()
2280 if view_cls in mounted:
2281 if _fr_globals.warn_on_misuse:
2282 warnings.warn(
2283 f"{view_cls.__name__} is already registered on this app/router; "
2284 f"this include_view() call is a no-op. Duplicate registration "
2285 f"usually means a double import or the decorator form combined "
2286 f"with an explicit include_view() call.",
2287 RestlyMisuseWarning,
2288 stacklevel=4,
2289 )
2290 return
2291 _prepare_view_class(view_cls)
2292 api_router = _init_api_router(view_cls)
2293 _register_for_resource_ref(parent_router, view_cls)
2294 parent_router.include_router(api_router)
2295 # Record only after the mount succeeded: a failed registration must not be
2296 # marked as mounted, or the next call would silently no-op a view whose
2297 # routes never made it onto the parent.
2298 mounted.add(view_cls)
2299 setattr(holder, "_fr_mounted_views", mounted)
2300 # Fallback registration for users who skip ``fr.configure(app=...)``.
2301 # ``register_default_exception_handlers`` is idempotent and only acts on
2302 # FastAPI apps (it ignores nested APIRouter parents).
2303 if isinstance(parent_router, fastapi.FastAPI):
2304 register_default_exception_handlers(parent_router)
2307#: Bare business-method names. A ``@route``-decorated method must not be
2308#: named like one of these: it would shadow the verb (which the ``handle_<verb>``
2309#: handlers call) and collide with ``<verb>_endpoint`` at the same path.
2310#: Override the business method *without* a decorator for domain logic; use
2311#: ``<verb>_endpoint`` or a distinct name for a custom route.
2312_BARE_VERB_NAMES = frozenset({"get_many", "get_one", "create", "update", "delete"})
2315def _reject_bare_verb_route_names(view_cls: type[View]) -> None:
2316 for name, value in view_cls.__dict__.items():
2317 if name in _BARE_VERB_NAMES and hasattr(value, "_api_route_args"):
2318 raise TypeError(
2319 f"{view_cls.__name__}.{name}() is a route method named like the "
2320 f"business method '{name}'. A route by that name shadows the "
2321 f"method and collides with the '{name}_endpoint' endpoint "
2322 f"method. Rename it to '{name}_endpoint' (to replace it) or "
2323 f"give the custom action its own name."
2324 )
2327#: The five endpoint methods defined by RestView / AsyncRestView.
2328_ENDPOINT_METHOD_NAMES = frozenset(
2329 {
2330 "get_many_endpoint",
2331 "get_one_endpoint",
2332 "create_endpoint",
2333 "update_endpoint",
2334 "delete_endpoint",
2335 }
2336)
2339def _warn_on_misuse(view_cls: type[View]) -> None:
2340 """Opt-in registration-time lint (``fr.configure(warn_on_misuse=True)``).
2342 Flags the dominant misuse patterns with the idiomatic fix named in
2343 each message. Heuristic, best-effort, and advisory: every pattern it flags
2344 has a legitimate use, so it warns (:class:`RestlyMisuseWarning`) rather
2345 than rejects. Must run *before* parent endpoints are copied into the
2346 subclass, while ``__dict__`` still holds only what the user wrote; only
2347 the registered class is linted, not user-defined intermediate bases.
2348 """
2349 own = view_cls.__dict__
2350 name = view_cls.__name__
2351 is_crud_view = issubclass(view_cls, BaseRestView)
2353 # 1. Endpoint-method override where a business-method override was likely meant.
2354 if is_crud_view:
2355 for endpoint in sorted(_ENDPOINT_METHOD_NAMES & own.keys()):
2356 verb = endpoint.removesuffix("_endpoint")
2357 warnings.warn(
2358 f"{name} overrides the endpoint method '{endpoint}'. Override "
2359 f"an endpoint method only to change the HTTP contract. For "
2360 f"domain logic override the business method '{verb}'; for a gate "
2361 f"'authorize'; for the response shape 'to_response'. Call "
2362 f"'handle_{verb}' from the replacement to keep the bracket.",
2363 RestlyMisuseWarning,
2364 stacklevel=5,
2365 )
2367 # 2. Manual session.commit() in a view method. The framework owns the
2368 # commit; methods that go through write_action / handle_<verb> are exempt.
2369 for attr, value in own.items():
2370 func = getattr(value, "__func__", value)
2371 if not isinstance(func, types.FunctionType):
2372 continue
2373 try:
2374 source = inspect.getsource(func)
2375 except (OSError, TypeError):
2376 continue
2377 if (
2378 ".commit(" in source
2379 and "write_action" not in source
2380 and "handle_" not in source
2381 ):
2382 warnings.warn(
2383 f"{name}.{attr} calls session.commit() directly. The framework "
2384 f"owns the commit: reuse handle_<verb>(), or bracket the "
2385 f"mutation with write_action('<action>', ...) so authorize / "
2386 f"before_action_commit / after_action_commit run.",
2387 RestlyMisuseWarning,
2388 stacklevel=5,
2389 )
2391 # 3. A CRUD route set hand-rolled on a bare View.
2392 if not is_crud_view:
2393 http_methods: set[str] = set()
2394 n_routes = 0
2395 for value in own.values():
2396 route_args = getattr(value, "_api_route_args", None)
2397 if route_args is None:
2398 continue
2399 _path, route_kwargs = route_args
2400 n_routes += 1
2401 http_methods.update(
2402 method.upper() for method in route_kwargs.get("methods", ["GET"])
2403 )
2404 if (
2405 n_routes >= 3
2406 and {"GET", "POST"} <= http_methods
2407 and http_methods & {"PATCH", "PUT", "DELETE"}
2408 ):
2409 warnings.warn(
2410 f"{name} hand-rolls a CRUD route set on a bare View. RestView / "
2411 f"AsyncRestView already define list/create/get/update/delete "
2412 f"endpoint methods. Subclass one and override the business methods "
2413 f"(create/update/delete), declare a scope, or override "
2414 f"authorize for custom behavior.",
2415 RestlyMisuseWarning,
2416 stacklevel=5,
2417 )
2419 # 4. A reference type (IDRef/IDSchema) whose name is a scalar FK column:
2420 # the ``post_id: IDRef[Post]`` mistake, where ``data.post_id`` becomes a
2421 # wrapper instead of the plain id.
2422 _warn_scalar_named_reference_fields(view_cls)
2425def _warn_scalar_named_reference_fields(view_cls: type[View]) -> None:
2426 """Flag an ``IDRef``/``IDSchema`` field whose name is a scalar foreign-key
2427 column: ``post_id: IDRef[Post]`` makes ``data.post_id`` a reference wrapper
2428 instead of the plain id. Steer to ``fr.MustExist[pk, Model]``. Best-effort:
2429 lints only an explicitly-declared ``schema`` against the view's ``model``,
2430 and never a relationship-named or list reference (those names aren't columns).
2431 """
2432 schema = view_cls.__dict__.get("schema")
2433 model = getattr(view_cls, "model", None)
2434 if schema is None or model is None:
2435 return
2436 try:
2437 mapper = sa_inspect(model)
2438 except Exception:
2439 return
2440 for field_name, field_info in schema.model_fields.items():
2441 origin_target = reference_origin_and_target(field_info.annotation)
2442 if origin_target is None:
2443 continue
2444 column = mapper.columns.get(field_name)
2445 if column is None or not column.foreign_keys:
2446 continue # a relationship-named or non-FK column is legitimate
2447 origin, target = origin_target
2448 ref_name = origin.__name__
2449 target_name = target.__name__ if target is not None else "Model"
2450 try:
2451 pk_name = column.type.python_type.__name__
2452 except Exception:
2453 pk_name = "int"
2454 rel = _relationship_name_for_fk(model, field_name)
2455 rename = (
2456 f", or name the field `{rel}` for the {ref_name} relationship"
2457 if rel
2458 else ""
2459 )
2460 warnings.warn(
2461 f"{schema.__name__}.{field_name} is typed `{ref_name}[{target_name}]`, "
2462 f"but `{field_name}` maps to the scalar foreign-key column "
2463 f"`{model.__name__}.{field_name}`. An {ref_name} field resolves to an "
2464 f"ORM object, so `data.{field_name}` is a reference wrapper, not the "
2465 f"plain id. Use `{field_name}: fr.MustExist[{pk_name}, {target_name}]` "
2466 f"for a checked scalar FK{rename}.",
2467 RestlyMisuseWarning,
2468 stacklevel=6,
2469 )
2472def _reject_buried_markers_in_view_schemas(view_cls: type[View]) -> None:
2473 """Backstop the ``BaseSchema`` import-time check at view registration.
2475 ``BaseSchema.__pydantic_init_subclass__`` already rejects a buried
2476 ReadOnly/WriteOnly marker as the schema class is defined, but a view may use
2477 a schema (and derived create/update schemas) that does not subclass
2478 ``BaseSchema``. Re-check the schemas this view actually uses so those are
2479 covered too.
2480 """
2481 checked: set[type] = set()
2482 for attr in ("schema", "schema_response", "schema_create", "schema_update"):
2483 schema = getattr(view_cls, attr, None)
2484 if schema is None or schema in checked:
2485 continue
2486 checked.add(schema)
2487 _reject_buried_markers(schema)
2490def _prepare_view_class(view_cls: type[View]) -> None:
2491 """Run the one-time class-level setup for a View.
2493 Guarded by the ``_fr_initialised`` marker (stored in ``__dict__`` so it is
2494 not inherited from a parent class that was registered separately). Calling
2495 this multiple times is a no-op after the first run.
2496 """
2497 if view_cls.__dict__.get("_fr_initialised", False):
2498 return
2499 if _fr_globals.warn_on_misuse:
2500 _warn_on_misuse(view_cls)
2501 _copy_all_parent_class_endpoints_into_this_subclass(view_cls)
2502 _reject_bare_verb_route_names(view_cls)
2503 _init_all_endpoints(view_cls)
2504 view_cls.before_include_view()
2505 _reject_buried_markers_in_view_schemas(view_cls)
2506 _init_class_based_view(view_cls)
2507 view_cls._fr_initialised = True # type: ignore[attr-defined]
2510def _copy_all_parent_class_endpoints_into_this_subclass(view_cls: type[View]):
2511 """
2512 Override all methods with a @route decorator of the parent classes of view_cls
2513 with a new copy directly on view_cls . This allows us to change the
2514 annotations on these endpoints without affecting the parent endpoints.
2516 For example, FooView.get() delegates to AsyncRestView.get() if it is not
2517 overridden (this is called implicit delegation through method resolution). And if
2518 we add the annotation that FooView.get() returns FooSchema but do not make a copy
2519 then AsyncRestView.get() and all other subclasses will get the FooSchema
2520 annotation as well.
2521 """
2522 for name, endpoint in _get_all_parent_endpoints(view_cls).items():
2523 # `name` is the attribute key (e.g. "get_many_endpoint"), which is stable
2524 # across copies; `endpoint.__name__` may have been mangled by a parent's
2525 # own registration (e.g. "parentview_get_many_endpoint"), so key off the
2526 # attribute name throughout.
2527 if name in view_cls.__dict__:
2528 # This endpoint is already overridden!
2529 continue
2531 # The original endpoint might be shared between subclasses.
2532 # So make a copy and put that on the view_cls.
2533 endpoint_wrapper = _make_copy(endpoint, view_cls)
2534 if getattr(endpoint, "__module__", "").startswith("fastapi_restly."):
2535 # The shells carry override-redirect docstrings for help()/source
2536 # readers. FastAPI reads endpoint.__doc__ as the OpenAPI operation
2537 # description, so strip the copy: framework guidance must not leak
2538 # into the user's API docs. User-defined endpoints keep theirs.
2539 endpoint_wrapper.__doc__ = None
2540 # Reset the copy's name to the endpoint attribute so downstream renaming
2541 # produces "<view>_<name>" even when the source was a parent's renamed
2542 # copy.
2543 endpoint_wrapper.__name__ = name
2544 # Set explicit __qualname__ for debugging purposes.
2545 endpoint_wrapper.__qualname__ = f"{view_cls.__name__}_{name}_wrapper"
2546 setattr(view_cls, name, endpoint_wrapper)
2549def _make_copy(endpoint: Callable, view_cls: type[View]) -> Callable:
2550 """
2551 Wrap the endpoint in a new function as kind of copy.
2553 Fun fact: You cannot do this inside a for loop, because the closure of 'endpoint'
2554 inside the wrapper works on the variable, not on the value. And for-loops in Python
2555 do not have their own variable scope.
2557 https://eev.ee/blog/2011/04/24/gotcha-python-scoping-closures/
2558 """
2559 if inspect.iscoroutinefunction(endpoint):
2561 @functools.wraps(endpoint)
2562 async def _async_wrapper(self, *args, **kwargs):
2563 return await endpoint(self, *args, **kwargs)
2565 endpoint_wrapper: Callable = _async_wrapper
2566 else:
2568 @functools.wraps(endpoint)
2569 def _sync_wrapper(self, *args, **kwargs):
2570 return endpoint(self, *args, **kwargs)
2572 endpoint_wrapper = _sync_wrapper
2574 endpoint_wrapper.__annotations__ = endpoint.__annotations__.copy()
2575 return endpoint_wrapper
2578def _init_all_endpoints(view_cls: type[View]):
2579 """
2580 Ensure every endpoint has a unique name and update the 'self' annotation.
2581 """
2582 for attr in view_cls.__dict__.values():
2583 if not hasattr(attr, "_api_route_args"):
2584 continue
2585 endpoint = attr
2586 # Give every endpoint a unique name
2587 # This will give the FooView.create() endpoint the name "fooview_create"
2588 endpoint.__name__ = view_cls.__name__.lower() + "_" + endpoint.__name__
2589 _annotate_self(view_cls, endpoint)
2592def _view_id_type(view_cls: type) -> Any:
2593 """The ``{id}`` type: ``id_type`` when set, else the model's primary key's.
2595 Resolved per registration and never stored on the class, so a subclass
2596 with another model derives its own.
2597 """
2598 explicit = getattr(view_cls, "id_type", None)
2599 if explicit is not None:
2600 return explicit
2601 model = getattr(view_cls, "model", None)
2602 # a composite key has no scalar type; it keeps the int default
2603 if model is None or len(sa_inspect(model).primary_key) != 1:
2604 return int
2605 return _model_id_type(model) or int
2608_UNCHANGED: Any = object()
2609_ANY_STRINGS = frozenset({"Any", "typing.Any"})
2612def _annotate(
2613 func: Callable,
2614 return_annotation: Any = _UNCHANGED,
2615 *,
2616 overwrite: bool = False,
2617 **param_annotations: Any,
2618) -> None:
2619 """Fill the view's concrete types into ``func.__signature__``.
2621 The endpoint shells declare ``Any`` because the concrete types only exist
2622 per view class. A parameter or return is filled only where the author
2623 wrote no annotation or ``Any``, so a replaced endpoint method keeps its
2624 own types. ``overwrite`` fills regardless.
2625 """
2626 # Judge what the author wrote, not the signature: functools.wraps copies
2627 # a registered parent's filled-in signature onto a subclass's copy.
2628 written = inspect.get_annotations(func)
2630 def fillable(name: str) -> bool:
2631 annotation = written.get(name, Any)
2632 # under PEP 563 the default signature's ``Any`` is the string "Any"
2633 return (
2634 overwrite
2635 or annotation is Any
2636 or (isinstance(annotation, str) and annotation in _ANY_STRINGS)
2637 )
2639 sig = inspect.signature(func)
2640 params = [
2641 param.replace(annotation=param_annotations[param.name])
2642 if param.name in param_annotations and fillable(param.name)
2643 else param
2644 for param in sig.parameters.values()
2645 ]
2646 if return_annotation is not _UNCHANGED and fillable("return"):
2647 sig = sig.replace(return_annotation=return_annotation)
2648 func.__signature__ = sig.replace(parameters=params) # type: ignore[attr-defined]
2651def _get_all_parent_endpoints(view_cls: type[View]) -> dict[str, Callable]:
2652 """Map parent route endpoints by attribute name.
2654 Registered intermediate parents copy base endpoints into their ``__dict__``.
2655 Keying by most-derived attribute name prevents duplicate logical endpoints
2656 while still respecting overrides.
2657 """
2658 endpoints: dict[str, Callable] = {}
2659 seen: set[str] = set()
2660 for cls in view_cls.mro():
2661 if cls is view_cls:
2662 continue
2663 for name, value in cls.__dict__.items():
2664 if name in seen:
2665 # A more-derived class already defined this name; it shadows the
2666 # base regardless of whether that override is itself a route.
2667 continue
2668 seen.add(name)
2669 if hasattr(value, "_api_route_args"):
2670 endpoints[name] = value
2671 return endpoints
2674def _init_api_router(view_cls: type[View]) -> fastapi.APIRouter:
2675 for name in view_cls.route_options:
2676 if not _is_route_name_in_lineage(view_cls, name):
2677 raise RestlyConfigurationError(
2678 f"{name!r} is not a route in {view_cls.__name__}.route_options"
2679 )
2681 # Each level of the class hierarchy adds to these, base first: a subclass
2682 # cannot drop a guard its base declares.
2683 levels = [c.__dict__ for c in reversed(view_cls.mro())]
2684 prefix = "".join(level["prefix"] for level in levels if "prefix" in level)
2685 dependencies: list[Any] = []
2686 responses: dict[int | str, dict[str, Any]] = {}
2687 for level in levels:
2688 for dependency in level.get("dependencies") or ():
2689 # A subclass that spells out ``[own, *Base.dependencies]`` lists
2690 # the base's entries again. They move to where it lists them, so
2691 # ``own`` still runs first.
2692 dependencies = [seen for seen in dependencies if seen is not dependency]
2693 dependencies.append(dependency)
2694 responses.update(level.get("responses") or {})
2695 tags = _get_router_tags(view_cls, prefix)
2696 api_router = fastapi.APIRouter(
2697 prefix=prefix, tags=tags, responses=responses, dependencies=dependencies
2698 )
2700 # Find all endpoint functions in this class and add them to the router
2701 list_parameters: list[dict[str, Any]] | None = None
2702 for name, attr in view_cls.__dict__.items():
2703 if not hasattr(attr, "_api_route_args"):
2704 continue
2705 endpoint = attr
2706 path, route_kwargs = endpoint._api_route_args
2707 defaults = {}
2708 if issubclass(view_cls, BaseRestView) and name in _CRUD_ROUTE_METADATA:
2709 route_name, summary = _CRUD_ROUTE_METADATA[name]
2710 # FastAPI carries placeholders through include_router(), so an
2711 # explicit generator on the route, router, or app still wins.
2712 # https://github.com/fastapi/fastapi/blob/0.115.0/fastapi/utils.py
2713 defaults = {
2714 "name": route_name,
2715 "summary": summary,
2716 "generate_unique_id_function": Default(
2717 functools.partial(
2718 _generate_crud_operation_id,
2719 is_member=name not in {"get_many_endpoint", "create_endpoint"},
2720 )
2721 ),
2722 }
2723 route_kwargs = {
2724 **defaults,
2725 **route_kwargs,
2726 **view_cls.route_options.get(name, {}),
2727 }
2728 if getattr(endpoint, "_fr_list_guard", False):
2729 if list_parameters is None:
2730 list_parameters = _list_openapi_parameters(
2731 cast(Any, view_cls).schema_list_params
2732 )
2733 extra = dict(route_kwargs.get("openapi_extra") or {})
2734 extra["parameters"] = [*list_parameters, *extra.get("parameters", [])]
2735 route_kwargs["openapi_extra"] = extra
2736 _add_api_route(api_router, view_cls, path, endpoint, route_kwargs)
2738 return api_router
2741_CRUD_ROUTE_METADATA = {
2742 "get_many_endpoint": ("get_many", "List"),
2743 "get_one_endpoint": ("get_one", "Get"),
2744 "create_endpoint": ("create", "Create"),
2745 "update_endpoint": ("update", "Update"),
2746 "delete_endpoint": ("delete", "Delete"),
2747 "put": ("put", "Update (PUT)"),
2748}
2751def _generate_crud_operation_id(
2752 route: fastapi.routing.APIRoute, *, is_member: bool
2753) -> str:
2754 """Combine the mounted resource path and action, omitting the member key."""
2755 path = route.path_format.rstrip("/")
2756 if is_member:
2757 path = re.sub(r"/\{[^}]+\}$", "", path)
2758 segments = [
2759 f"by_{segment[1:-1]}" if segment.startswith("{") else segment
2760 for segment in path.strip("/").split("/")
2761 if segment
2762 ]
2763 resource = re.sub(r"\W+", "_", "_".join(segments)).strip("_") or "root"
2764 action = {"get_many": "list", "get_one": "get"}.get(route.name, route.name)
2765 return re.sub(r"\W+", "_", f"{resource}_{action}").strip("_")
2768def _get_router_tags(view_cls: type[View], prefix: str) -> list[str | Enum]:
2769 if view_cls.tags is not None:
2770 return list(view_cls.tags)
2771 return [_derive_tag_from_prefix(prefix) or view_cls.__name__]
2774def _derive_tag_from_prefix(prefix: str) -> str | None:
2775 segments = [segment for segment in prefix.strip("/").split("/") if segment]
2776 if not segments:
2777 return None
2778 return segments[-1].replace("-", " ").replace("_", " ").title()
2781def _add_api_route(
2782 api_router: fastapi.APIRouter,
2783 view_cls: type[View],
2784 path: str,
2785 endpoint: Callable,
2786 route_kwargs: dict[str, Any],
2787) -> None:
2788 if _should_add_collection_route_alias(view_cls, path):
2789 api_router.add_api_route("", endpoint, **route_kwargs)
2790 hidden_alias_kwargs = {**route_kwargs, "include_in_schema": False}
2791 api_router.add_api_route("/", endpoint, **hidden_alias_kwargs)
2792 return
2794 api_router.add_api_route(path, endpoint, **route_kwargs)
2797def _should_add_collection_route_alias(view_cls: type[View], path: str) -> bool:
2798 return issubclass(view_cls, BaseRestView) and path == "/"
2801def _annotate_self(view_cls: type[View], endpoint: Callable) -> None:
2802 """
2803 Annotate the 'self' argument as 'self=Depends(view_cls)'. That way FastAPI instantiates the
2804 view_cls before calling the endpoint function and passes it as 'self'.
2805 Note that it sets endpoint.__signature__ which overrides any other inspection.
2807 Note: Copied (MIT license) and adjusted from: https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py
2809 Fixes the endpoint signature to ensure FastAPI performs dependency injection properly.
2810 """
2811 sig = inspect.signature(endpoint)
2812 params: list[inspect.Parameter] = list(sig.parameters.values())
2813 self_param = params[0]
2814 new_self_param = self_param.replace(default=fastapi.Depends(view_cls))
2816 new_params = [new_self_param] + [
2817 param.replace(kind=inspect.Parameter.KEYWORD_ONLY) for param in params[1:]
2818 ]
2819 endpoint.__signature__ = sig.replace(parameters=new_params) # type: ignore[attr-defined]
2822# Bare-typed annotations FastAPI special-cases for parameter injection
2823# (no ``Depends(...)`` marker required). Treated alongside ``Depends``-
2824# marked annotations as DI-wired class attributes; everything else is
2825# left as plain typing.
2826_FASTAPI_SPECIAL_INJECTABLE: tuple[type, ...] = (
2827 Request,
2828 Response,
2829 BackgroundTasks,
2830 WebSocket,
2831)
2833# Markers for the request inputs FastAPI reads into an endpoint parameter:
2834# ``Path``/``Query``/``Header``/``Cookie`` (``Param``) and ``Body``/``Form``/
2835# ``File`` (``Body``). FastAPI fills a class attribute only through ``Depends``
2836# or a special type, never through one of these, so the view is rejected.
2837_REQUEST_PARAMETER_MARKERS: tuple[type, ...] = (_ParamMarker, _BodyMarker)
2840def _init_class_based_view(view_cls: type[View]) -> None:
2841 """
2842 Note: Copied (MIT license) and adjusted from: https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/cbv.py
2844 Idempotently modifies the provided `cls`, performing the following modifications:
2845 * The `__init__` function is updated to set any class-annotated dependencies as instance attributes
2846 * The `__signature__` attribute is updated to indicate to FastAPI what arguments should be passed to the initializer
2847 """
2848 if getattr(view_cls, "__class_based_view", False):
2849 return # Already initialized
2850 old_init: Callable[..., Any] = view_cls.__init__
2851 old_signature = inspect.signature(old_init)
2852 old_parameters = list(old_signature.parameters.values())[1:] # drop `self`
2853 new_parameters = [
2854 x
2855 for x in old_parameters
2856 if x.kind
2857 not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
2858 ]
2859 # Marker-based DI with MRO-aware shadowing: walk the MRO from the
2860 # base classes upward and pick, for each name, an annotation that
2861 # either carries a ``Depends(...)`` marker or names one of FastAPI's
2862 # bare-injectable special types (``Request`` / ``Response`` etc.).
2863 # A *plain* annotation on a more-derived class (e.g. a mixin
2864 # declaring ``session: AsyncSession`` for static-typing purposes)
2865 # does NOT shadow a marker-bearing annotation from a base — the
2866 # framework prefers wiring fidelity over the most-derived hint.
2867 # Without this rule, any plain annotation a mixin adds would
2868 # silently break dependency injection.
2869 di_annotations: dict[str, Any] = {}
2870 request_markers: dict[str, Any] = {}
2871 for cls in reversed(view_cls.__mro__):
2872 try:
2873 cls_hints = get_type_hints(cls, include_extras=True)
2874 except Exception:
2875 continue
2876 for name, annotation in cls_hints.items():
2877 if get_origin(annotation) is ClassVar:
2878 continue
2879 metadata = getattr(annotation, "__metadata__", ())
2880 has_depends_marker = any(isinstance(m, _DependsMarker) for m in metadata)
2881 for marker in metadata:
2882 if isinstance(marker, _REQUEST_PARAMETER_MARKERS):
2883 request_markers.setdefault(name, marker)
2884 underlying = (
2885 annotation
2886 if get_origin(annotation) is not Annotated
2887 else (get_args(annotation)[0] if get_args(annotation) else annotation)
2888 )
2889 # On Python 3.10, a parameterized generic such as list[str] passes
2890 # isclass() but makes issubclass() raise; it is never special.
2891 is_special_type = (
2892 inspect.isclass(underlying)
2893 and get_origin(underlying) is None
2894 and issubclass(underlying, _FASTAPI_SPECIAL_INJECTABLE)
2895 )
2896 if has_depends_marker or is_special_type:
2897 # Marker-bearing annotation wins, regardless of MRO position.
2898 di_annotations[name] = annotation
2899 # Plain annotations are silently ignored — they neither set
2900 # nor clear an entry in di_annotations.
2902 # FastAPI never fills an attribute with a request-parameter marker: it
2903 # would keep its class default, with no error and no OpenAPI entry. A
2904 # Depends marker on the same name wires it, as above.
2905 for name, marker in request_markers.items():
2906 if name not in di_annotations: 2906 ↛ 2905line 2906 didn't jump to line 2905 because the condition on line 2906 was always true
2907 raise RestlyConfigurationError(
2908 f"{view_cls.__name__}.{name} has a {type(marker).__name__}() "
2909 "marker, but a view class attribute is filled only through "
2910 "Depends(...) or Security(...), so FastAPI would never set it. "
2911 "Declare the parameter on the endpoint method that reads it, "
2912 f"or annotate {name} with Annotated[..., Depends(...)] and read "
2913 "the parameter in that dependency."
2914 )
2916 dependency_names: list[str] = []
2917 for name, annotation in di_annotations.items():
2918 dependency_names.append(name)
2919 default_value = getattr(view_cls, name, inspect.Parameter.empty)
2920 new_parameters.append(
2921 inspect.Parameter(
2922 name=name,
2923 kind=inspect.Parameter.KEYWORD_ONLY,
2924 default=default_value,
2925 annotation=annotation,
2926 )
2927 )
2928 new_signature = old_signature.replace(parameters=new_parameters)
2930 def new_init(self: Any, *args: Any, **kwargs: Any) -> None:
2931 for dep_name in dependency_names:
2932 dep_value = kwargs.pop(dep_name)
2933 setattr(self, dep_name, dep_value)
2934 old_init(self, *args, **kwargs)
2936 setattr(view_cls, "__signature__", new_signature)
2937 setattr(view_cls, "__init__", new_init)
2938 setattr(view_cls, "__class_based_view", True)