Coverage for fastapi_restly / views / _react_admin.py: 85%
182 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-24 11:13 +0000
1"""
2React Admin compatible views for fastapi-restly.
4Implements the ra-data-simple-rest wire contract for list:
5- response body: plain JSON array
6- sort: sort=["field","ASC|DESC"]
7- range: range=[start, end] (both inclusive, e.g. [0,24] = 25 items)
8- filter: filter={"field":"value"} or filter={"id":[1,2,3]} for getMany
9- Content-Range: items 0-24/315
10"""
12import json
13from dataclasses import dataclass
14from typing import Any, ClassVar, Protocol, Sequence, cast
16import fastapi
17import pydantic
18import sqlalchemy
19from sqlalchemy.orm import DeclarativeBase, RelationshipProperty
21from ..exc import BadQueryParam
22from ._async import AsyncRestView
23from ._base import ListingResult, ResponseShape, _annotate, get, put
24from ._sync import RestView
26#: Default page size used when the react-admin client does not send a `range`
27#: query parameter. Override per-view via ``default_page_size``.
28DEFAULT_REACT_ADMIN_PAGE_SIZE = 25
31@dataclass(frozen=True)
32class _ReactAdminListParams:
33 sort: tuple[str, str] | None
34 start: int
35 end: int
36 filters: dict
39# ---------------------------------------------------------------------------
40# Query parsing helpers (standalone, analogous to query/_impl.py)
41# ---------------------------------------------------------------------------
44def parse_react_admin_sort(sort_raw: str | None) -> tuple[str, str] | None:
45 """
46 Parse a react-admin sort query parameter.
48 Expected: '["field","ASC"]' or '["field","DESC"]'
49 Returns (field, direction) or None if absent.
50 Raises HTTPException 400 on malformed input.
51 """
52 if not sort_raw:
53 return None
54 try:
55 parsed = json.loads(sort_raw)
56 except json.JSONDecodeError:
57 raise BadQueryParam("Invalid sort parameter: must be a JSON array")
58 if not isinstance(parsed, list) or len(parsed) != 2: 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true
59 raise fastapi.HTTPException(
60 400, "Invalid sort parameter: must be [field, direction]"
61 )
62 field, direction = parsed
63 if not isinstance(field, str) or direction not in ("ASC", "DESC"): 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true
64 raise fastapi.HTTPException(
65 400, "Invalid sort parameter: direction must be 'ASC' or 'DESC'"
66 )
67 return field, direction
70def parse_react_admin_range(
71 range_raw: str | None, default_page_size: int = DEFAULT_REACT_ADMIN_PAGE_SIZE
72) -> tuple[int, int]:
73 """
74 Parse a react-admin range query parameter.
76 Expected: '[0,24]' (both inclusive).
77 Returns (start, end). Defaults to (0, default_page_size - 1) if absent.
78 Raises HTTPException 400 on malformed input.
79 """
80 if not range_raw:
81 return 0, default_page_size - 1
82 try:
83 parsed = json.loads(range_raw)
84 except json.JSONDecodeError:
85 raise fastapi.HTTPException(
86 400, "Invalid range parameter: must be a JSON array"
87 )
88 if not isinstance(parsed, list) or len(parsed) != 2: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true
89 raise fastapi.HTTPException(
90 400, "Invalid range parameter: must be [start, end]"
91 )
92 start, end = parsed
93 if not isinstance(start, int) or not isinstance(end, int): 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 raise fastapi.HTTPException(
95 400, "Invalid range parameter: values must be integers"
96 )
97 return start, end
100def parse_react_admin_filter(filter_raw: str | None) -> dict:
101 """
102 Parse a react-admin filter query parameter.
104 Expected: '{"field":"value"}' or '{"id":[1,2,3]}' for getMany.
105 Returns a dict. Defaults to {} if absent.
106 Raises HTTPException 400 on malformed input.
107 """
108 if not filter_raw:
109 return {}
110 try:
111 parsed = json.loads(filter_raw)
112 except json.JSONDecodeError:
113 raise fastapi.HTTPException(
114 400, "Invalid filter parameter: must be a JSON object"
115 )
116 if not isinstance(parsed, dict): 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 raise fastapi.HTTPException(
118 400, "Invalid filter parameter: must be a JSON object"
119 )
120 return parsed
123def _resolve_column(
124 model: type[DeclarativeBase], schema_cls: Any, field_name: str
125) -> Any:
126 """
127 Resolve a PUBLIC schema field name (or alias) to its SQLAlchemy column.
129 Strict: only fields exposed on the response schema may be filtered or
130 sorted. A column that exists on the model but is omitted from the schema
131 (or marked write-only) is rejected, so the list endpoint cannot be used as
132 an oracle to filter/sort on -- and thereby probe -- hidden data. Mirrors the
133 standard REST dialect's schema-driven resolution.
135 Raises HTTPException 400 if the field is not a public, filterable schema field.
136 """
137 from ..schemas._base import is_writeonly_field
139 resolved_name: str | None = None
140 if schema_cls is not None: 140 ↛ 148line 140 didn't jump to line 148 because the condition on line 140 was always true
141 for name, field in schema_cls.model_fields.items():
142 if is_writeonly_field(schema_cls, name): 142 ↛ 143line 142 didn't jump to line 143 because the condition on line 142 was never true
143 continue
144 if name == field_name or field.alias == field_name:
145 resolved_name = name
146 break
148 if resolved_name is None:
149 raise BadQueryParam(f"Unknown filter field: {field_name!r}")
151 col = getattr(model, resolved_name, None)
152 if col is None: 152 ↛ 153line 152 didn't jump to line 153 because the condition on line 152 was never true
153 raise BadQueryParam(f"Unknown filter field: {field_name!r}")
154 if hasattr(col, "property") and isinstance(col.property, RelationshipProperty): 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 raise fastapi.HTTPException(
156 400, f"Cannot sort or filter by relationship field: {field_name!r}"
157 )
158 return col
161def _coerce_value(col: Any, value: Any) -> Any:
162 """Coerce a filter value to the column's Python type if needed.
164 Handles cases such as UUID strings that must be converted to uuid.UUID
165 objects before being passed to SQLAlchemy's type processor.
166 """
167 try:
168 py_type = col.type.python_type
169 except NotImplementedError:
170 return value
171 if not isinstance(value, py_type): 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true
172 try:
173 return py_type(value)
174 except (ValueError, TypeError):
175 raise fastapi.HTTPException(
176 400, f"Invalid filter value for {col.key!r}: {value!r}"
177 )
178 return value
181def _apply_react_admin_filters(
182 query: sqlalchemy.Select,
183 model: type[DeclarativeBase],
184 schema_cls: Any,
185 filters: dict,
186) -> sqlalchemy.Select:
187 """Apply a react-admin filter dict to a select query."""
188 for key, value in filters.items():
189 col = _resolve_column(model, schema_cls, key)
190 if isinstance(value, list):
191 coerced = [_coerce_value(col, v) for v in value]
192 query = query.where(col.in_(coerced))
193 else:
194 query = query.where(col == _coerce_value(col, value))
195 return query
198def apply_react_admin_query(
199 query: sqlalchemy.Select,
200 model: type[DeclarativeBase],
201 schema_cls: Any,
202 sort: tuple[str, str] | None,
203 start: int,
204 end: int,
205 filters: dict,
206) -> sqlalchemy.Select:
207 """
208 Apply filter, sort, and range (limit/offset) to a select query.
210 This is the main query transformation entry point, analogous to
211 :func:`fastapi_restly.query.apply_list_params` for the standard REST dialect.
212 """
213 query = _apply_react_admin_filters(query, model, schema_cls, filters)
215 id_col = getattr(model, "id", None)
216 if sort:
217 field, direction = sort
218 col = _resolve_column(model, schema_cls, field)
219 order_fn = sqlalchemy.desc if direction == "DESC" else sqlalchemy.asc
220 query = query.order_by(order_fn(col))
221 # Append the PK as a final tiebreaker so pagination is deterministic on
222 # a non-unique sort column (mirrors _apply_sorting in query/_impl.py).
223 if id_col is not None and col is not id_col:
224 query = query.order_by(id_col)
225 elif id_col is not None: 225 ↛ 228line 225 didn't jump to line 228 because the condition on line 225 was always true
226 query = query.order_by(id_col)
228 query = query.limit(end - start + 1).offset(start)
229 return query
232# ---------------------------------------------------------------------------
233# Shared implementation mixin
234# ---------------------------------------------------------------------------
237class _ReactAdminViewProtocol(Protocol):
238 request: fastapi.Request
239 model: ClassVar[type[DeclarativeBase]]
240 schema: ClassVar[type[pydantic.BaseModel]]
241 schema_update: ClassVar[type[pydantic.BaseModel]]
242 id_type: ClassVar[type[Any]]
243 default_page_size: ClassVar[int | None]
244 get_many_endpoint: ClassVar[Any]
245 put: ClassVar[Any]
247 def get_react_admin_range_unit(self) -> str: ...
248 def get_relationship_loader_options(self) -> list[Any]: ...
249 def to_response_schema(self, obj: Any) -> pydantic.BaseModel: ...
250 def build_query(self) -> sqlalchemy.Select: ...
252 @classmethod
253 def before_include_view(cls) -> None: ...
256class _ReactAdminMixin:
257 """
258 Shared transport helpers for react-admin views.
260 This is an implementation detail shared by ReactAdminView and
261 AsyncReactAdminView. User-facing customization should happen by
262 subclassing one of those concrete view classes.
264 Set :attr:`default_page_size` on a subclass to change the implicit page
265 size used when the client does not send a ``range`` query parameter.
267 Type annotations on the mixin methods use ``_ReactAdminViewProtocol`` to
268 make the expected ``RestView`` surface explicit to static checkers.
269 """
271 #: Implicit page size when no ``range`` parameter is sent. Override per-view.
272 default_page_size: ClassVar[int | None] = DEFAULT_REACT_ADMIN_PAGE_SIZE
274 def get_react_admin_range_unit(self) -> str:
275 """Return the unit string used in the Content-Range header."""
276 return "items"
278 def _parse_react_admin_params(self) -> _ReactAdminListParams:
279 """Parse sort, range, and filter from the current request query string."""
280 view = cast(_ReactAdminViewProtocol, self)
281 return self._coerce_react_admin_params(view.request.query_params)
283 def _coerce_react_admin_params(self, params: Any) -> _ReactAdminListParams:
284 if isinstance(params, _ReactAdminListParams):
285 return params
287 view = cast(_ReactAdminViewProtocol, self)
288 default_page_size = view.default_page_size or DEFAULT_REACT_ADMIN_PAGE_SIZE
289 sort_raw = params.get("sort") if hasattr(params, "get") else None
290 range_raw = params.get("range") if hasattr(params, "get") else None
291 filter_raw = params.get("filter") if hasattr(params, "get") else None
293 sort = (
294 tuple(sort_raw)
295 if isinstance(sort_raw, list | tuple)
296 else parse_react_admin_sort(sort_raw)
297 )
298 if isinstance(range_raw, list | tuple): 298 ↛ 299line 298 didn't jump to line 299 because the condition on line 298 was never true
299 if len(range_raw) != 2:
300 raise fastapi.HTTPException(
301 400, "Invalid range parameter: must be [start, end]"
302 )
303 start, end = range_raw
304 else:
305 start, end = parse_react_admin_range(
306 range_raw, default_page_size=default_page_size
307 )
308 filters = (
309 filter_raw
310 if isinstance(filter_raw, dict)
311 else parse_react_admin_filter(filter_raw)
312 )
313 return _ReactAdminListParams(sort=sort, start=start, end=end, filters=filters)
315 def apply_query_params(
316 self, query: sqlalchemy.Select, query_params: Any
317 ) -> sqlalchemy.Select:
318 """Apply the react-admin list grammar to the scoped base query."""
319 view = cast(_ReactAdminViewProtocol, self)
320 params = self._coerce_react_admin_params(query_params)
321 return apply_react_admin_query(
322 query,
323 view.model,
324 view.schema,
325 params.sort,
326 params.start,
327 params.end,
328 params.filters,
329 )
331 def _serialize_items(self, items: Sequence[Any]) -> list[dict]:
332 """Serialize ORM objects to JSON-compatible dicts via the view's response schema."""
333 view = cast(_ReactAdminViewProtocol, self)
334 return [
335 view.to_response_schema(obj).model_dump(mode="json", by_alias=True)
336 for obj in items
337 ]
339 def _build_react_admin_list_response(
340 self, serialized_items: list[dict], total: int, start: int, end: int
341 ) -> fastapi.Response:
342 """Build a JSON array response with a Content-Range header."""
343 view = cast(_ReactAdminViewProtocol, self)
344 unit = view.get_react_admin_range_unit()
345 last = start + len(serialized_items) - 1 if serialized_items else start
346 return fastapi.Response(
347 content=json.dumps(serialized_items),
348 media_type="application/json",
349 headers={
350 "Content-Range": f"{unit} {start}-{last}/{total}",
351 "Access-Control-Expose-Headers": "Content-Range",
352 },
353 )
355 def to_react_admin_listing_response(
356 self, listing_result: ListingResult[Any]
357 ) -> fastapi.Response:
358 params = self._coerce_react_admin_params(listing_result.query_params)
359 return self._build_react_admin_list_response(
360 self._serialize_items(listing_result.objects),
361 listing_result.total_count,
362 params.start,
363 params.end,
364 )
366 def to_response(
367 self, obj_or_list: Any, shape: ResponseShape = ResponseShape.SINGLE
368 ) -> Any:
369 if shape is ResponseShape.LISTING:
370 return self.to_react_admin_listing_response(obj_or_list)
371 return cast(Any, super()).to_response(obj_or_list, shape)
373 @classmethod
374 def before_include_view(cls) -> None:
375 cast(Any, super()).before_include_view()
376 view_cls = cast(type[_ReactAdminViewProtocol], cls)
377 # Override the list return annotation set by BaseRestView to Response,
378 # since we return a raw Response with Content-Range header.
379 if hasattr(view_cls, "get_many_endpoint"): 379 ↛ 382line 379 didn't jump to line 382 because the condition on line 379 was always true
380 _annotate(view_cls.get_many_endpoint, return_annotation=fastapi.Response)
381 # Annotate the PUT handler with the same schema/types as PATCH.
382 if hasattr(view_cls, "put"): 382 ↛ exitline 382 didn't return from function 'before_include_view' because the condition on line 382 was always true
383 _annotate(
384 view_cls.put,
385 return_annotation=view_cls.schema,
386 schema_obj=view_cls.schema_update,
387 id=view_cls.id_type,
388 )
391# ---------------------------------------------------------------------------
392# Concrete view classes
393# ---------------------------------------------------------------------------
396class AsyncReactAdminView(_ReactAdminMixin, AsyncRestView):
397 """
398 AsyncRestView that speaks the ra-data-simple-rest wire contract.
400 Use this instead of AsyncRestView when your frontend is react-admin
401 with ra-data-simple-rest.
402 """
404 @get("/")
405 async def get_many_endpoint(self) -> Any:
406 result = await self.handle_get_many(self._parse_react_admin_params())
407 return self.to_response(result, ResponseShape.LISTING)
409 @put("/{id}")
410 async def put(self, id: Any, schema_obj: Any) -> Any:
411 obj = await self.handle_update(id, schema_obj)
412 return self.to_response(obj)
415class ReactAdminView(_ReactAdminMixin, RestView):
416 """
417 RestView that speaks the ra-data-simple-rest wire contract.
419 Use this instead of RestView when your frontend is react-admin
420 with ra-data-simple-rest.
421 """
423 @get("/")
424 def get_many_endpoint(self) -> Any:
425 result = self.handle_get_many(self._parse_react_admin_params())
426 return self.to_response(result, ResponseShape.LISTING)
428 @put("/{id}")
429 def put(self, id: Any, schema_obj: Any) -> Any:
430 obj = self.handle_update(id, schema_obj)
431 return self.to_response(obj)