Coverage for fastapi_restly / query / _impl.py: 97%

1import collections.abc as _abc 

2import datetime as _dt 

3import decimal as _decimal 

4import functools 

5import uuid as _uuid 

6import weakref 

7from collections import defaultdict 

8from typing import ( 

9 Annotated, 

10 Any, 

11 Callable, 

12 Iterator, 

13 Optional, 

14 cast, 

15 get_args, 

16 get_origin, 

17) 

18 

19import pydantic 

20import sqlalchemy 

21from pydantic import Field 

22from pydantic.fields import FieldInfo 

23from pydantic_core import SchemaValidator, core_schema 

24from sqlalchemy import ColumnElement, Select 

25from sqlalchemy.orm import DeclarativeBase, aliased 

26from sqlalchemy.orm.attributes import InstrumentedAttribute 

27from sqlalchemy.orm.properties import ColumnProperty 

28from sqlalchemy.orm.util import AliasedClass 

29from starlette.datastructures import QueryParams 

30 

31from .._pagination import ( 

32 _DEFAULT_PAGINATION, 

33 _SORT_QUERY_PARAM, 

34 NoPagination, 

35 NumberedPagination, 

36) 

37from ..exc import BadQueryParam 

38from ..schemas._base import ( 

39 IDRef, 

40 IDSchema, 

41 _schema_role_name, 

42 _unwrap_optional_annotation, 

43) 

44from ._shared import _append_pk_tiebreak, _escape_like_value 

45 

46SchemaType = type[pydantic.BaseModel] 

47_JoinAliases = dict[tuple[str, ...], AliasedClass[Any]] 

48 

49_AnyPagination = NumberedPagination | NoPagination | None 

50 

51#: The pagination that each generated list params model was created with, so 

52#: :func:`apply_list_params` and view registration can read it back. 

53_CREATED_WITH: "weakref.WeakKeyDictionary[type, _AnyPagination]" = ( 

54 weakref.WeakKeyDictionary() 

55) 

56 

57 

58class _FromParams: 

59 """The default of :func:`apply_list_params`'s ``pagination``: the one its 

60 params model was created with.""" 

61 

62 def __repr__(self) -> str: 

63 # the default as the API reference shows it; the docstring explains it 

64 return "..." 

65 

66 

67_FROM_PARAMS = _FromParams() 

68_UNKNOWN = object() 

69 

70 

71def _created_with(params_model: type) -> Any: 

72 """The pagination that generated list params were created with, or 

73 ``_UNKNOWN`` for hand-written ones.""" 

74 return _CREATED_WITH.get(params_model, _UNKNOWN) 

75 

76 

77def _query_settings(pagination: _AnyPagination) -> tuple[Any, ...] | None: 

78 """What the pagination fields of list params depend on; ``None`` for 

79 no pagination.""" 

80 if not isinstance(pagination, NumberedPagination): 

81 return None 

82 return ( 

83 pagination.page_query_param, 

84 pagination.page_size_query_param, 

85 pagination.default_page_size, 

86 pagination.max_page_size, 

87 pagination.max_page, 

88 ) 

89 

90 

91#: Largest pagination offset accepted by supported SQL databases. SQLite and 

92#: PostgreSQL bind ``OFFSET`` as a signed 64-bit integer. 

93_MAX_PAGINATION_OFFSET = 2**63 - 1 

94 

95 

96def _reserved_names(pagination: _AnyPagination) -> frozenset[str]: 

97 """Query parameters the schema produces besides filters. 

98 

99 A filter column whose public name is one of these would shadow pagination 

100 or sort, which would silently break the endpoint contract. Treated as a 

101 hard error. 

102 """ 

103 names = {_SORT_QUERY_PARAM} 

104 if isinstance(pagination, NumberedPagination): 

105 names |= {pagination.page_query_param, pagination.page_size_query_param} 

106 return frozenset(names) 

107 

108 

109# Types that support SQL ``<``/``<=``/``>``/``>=`` comparisons. Booleans are 

110# excluded: ordering booleans is rarely meaningful, and ``WHERE active >= true`` 

111# raises ``sqlalchemy.exc.ArgumentError`` at query time, which would otherwise 

112# surface to the client as a 500. 

113_ORDERABLE_TYPES: tuple[type, ...] = ( 

114 int, 

115 float, 

116 _decimal.Decimal, 

117 _dt.date, 

118 _dt.datetime, 

119 _dt.time, 

120 _dt.timedelta, 

121 str, 

122) 

123 

124 

125def _is_string_field(field: FieldInfo) -> bool: 

126 annotation = _unwrap_optional_annotation(field.annotation) 

127 return annotation is str 

128 

129 

130def _is_idref_field(field: FieldInfo) -> bool: 

131 """True for a scalar ``IDRef[T]`` FK field (e.g. ``post_id: IDRef[Post]``). 

132 

133 That ``*_id``-named spelling is discouraged -- the misuse lint steers it to 

134 ``fr.MustExist[int, Post]`` -- but the query layer still handles it here so 

135 the field keeps its filter param. 

136 

137 ``IDRef`` is a ``BaseModel`` subclass, so without this it would be recursed 

138 into as a nested schema and yield only ``post_id.id`` -- which never resolves 

139 (``post_id`` is a scalar column, not a relationship), leaving the FK with no 

140 filter param at all. Treated as a leaf, it filters on its own public name. 

141 Targets ``IDRef`` specifically, not ``IDSchema``: a nested *resource* schema 

142 that embeds its ``id`` also subclasses ``IDSchema`` and must keep its dotted 

143 traversal. ``list[IDRef[T]]`` (to-many) is unaffected -- its annotation is a 

144 list, so it is never a nested schema here. 

145 """ 

146 annotation = _unwrap_optional_annotation(field.annotation) 

147 return isinstance(annotation, type) and issubclass(annotation, IDRef) 

148 

149 

150def _supports_scalar_operators(field: FieldInfo) -> bool: 

151 """False for collection-typed fields (``dict``, ``list[str]``, ``Sequence``, 

152 ...) that no query-string value can coerce into: ``eq``/``__in``/``__ne`` 

153 (and range) filters on such a field would always 400 at request time, so 

154 only ``__isnull`` is generated -- the same advertise-only-what-works 

155 principle applied to non-column fields. 

156 

157 The check mirrors what the request path's ``_parse_value`` (full Pydantic 

158 field validation) will accept. ``Annotated`` wrappers are unwrapped even 

159 inside ``Optional`` (``conlist(...)`` produces those), and a 

160 ``pydantic.Json[...]`` field counts as scalar: its validation parses the 

161 query string as JSON, so its filters work. ``str``/``bytes`` are 

162 collections in the abc sense but coerce fine, so they stay scalar. 

163 """ 

164 annotation = field.annotation 

165 metadata: list[Any] = list(field.metadata) 

166 while True: 

167 annotation = _unwrap_optional_annotation(annotation) 

168 if get_origin(annotation) is Annotated: 

169 annotation, *extra = get_args(annotation) 

170 metadata.extend(extra) 

171 continue 

172 break 

173 # ``pydantic.Json`` is a real class at runtime; the type stubs model it as 

174 # an ``Annotated`` factory, hence the cast. 

175 json_marker = cast(type, pydantic.Json) 

176 if any(isinstance(item, json_marker) for item in metadata): 

177 return True 

178 origin = get_origin(annotation) or annotation 

179 return not ( 

180 isinstance(origin, type) 

181 and issubclass(origin, _abc.Collection) 

182 and not issubclass(origin, (str, bytes)) 

183 ) 

184 

185 

186def _supports_range_operators(field: FieldInfo) -> bool: 

187 annotation = _unwrap_optional_annotation(field.annotation) 

188 if annotation is bool: 

189 return False 

190 if not isinstance(annotation, type): 

191 return True 

192 if issubclass(annotation, bool): 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true

193 return False 

194 if issubclass(annotation, _ORDERABLE_TYPES): 

195 return True 

196 if issubclass(annotation, _uuid.UUID): 

197 return False 

198 return False 

199 

200 

201def _pagination_offset_validator(pagination: NumberedPagination) -> Any: 

202 """A model validator that rejects a page whose SQL offset would not fit in 

203 a signed 64-bit integer.""" 

204 page_param = pagination.page_query_param 

205 page_size_param = pagination.page_size_query_param 

206 

207 @pydantic.model_validator(mode="after") 

208 def _validate_pagination_offset(params: pydantic.BaseModel) -> pydantic.BaseModel: 

209 page = cast(int, getattr(params, page_param)) 

210 page_size = cast(Optional[int], getattr(params, page_size_param)) 

211 if page_size is None: 211 ↛ 212line 211 didn't jump to line 212 because the condition on line 211 was never true

212 return params 

213 if page - 1 > _MAX_PAGINATION_OFFSET // page_size: 

214 raise ValueError( 

215 f"{page_param} and {page_size_param} produce an offset above " 

216 f"{_MAX_PAGINATION_OFFSET}" 

217 ) 

218 return params 

219 

220 return _validate_pagination_offset 

221 

222 

223def derive_schema_list_params( 

224 schema_response: SchemaType, 

225 model: type[DeclarativeBase], 

226 *, 

227 pagination: _AnyPagination = _DEFAULT_PAGINATION, 

228) -> SchemaType: 

229 """ 

230 Build the list params: a Pydantic model that describes and validates the 

231 URL query parameters of a list endpoint. A view keeps it in 

232 :attr:`~fastapi_restly.views.BaseRestView.schema_list_params`. 

233 

234 The generated model accepts pagination (``page``, ``page_size``), sorting 

235 (``sort``), and one filter parameter per field of ``schema_response`` that maps to 

236 a filterable column on ``model`` -- with optional ``__in``/``__ne``/``__gte``/ 

237 ``__lte``/``__gt``/``__lt``/``__isnull``/``__contains``/``__icontains`` 

238 suffixes. Fields that do not resolve to a column (relationship/collection 

239 fields, or reference traversals the request path would reject) get no filter 

240 params, so the generated schema -- and the OpenAPI it produces -- no longer 

241 advertises filters for fields that are not filterable at all. WriteOnly 

242 fields get none either, since a filter on one would leak its value. Fields 

243 whose type is a collection (``dict``/``list``, e.g. ``JSON`` or ``ARRAY`` 

244 columns) generate only ``__isnull``: a query-string value cannot coerce into 

245 them, so every other operator would fail at request time. 

246 

247 With a ``pagination`` (the default is ``page`` and ``page_size``), its two 

248 query parameters are added under its names and validated by Pydantic with 

249 bounds: the page at least 1 and at most ``max_page`` when set, the page 

250 size from 1 to ``max_page_size``. The resulting SQL offset must also fit in 

251 a signed 64-bit integer. Out-of-range values produce a standard 422 

252 response from FastAPI. With a :class:`~fastapi_restly.views.NoPagination` 

253 or ``None``, no pagination parameters are emitted at all -- the endpoint 

254 returns every matching row -- while sorting and filtering stay available. 

255 

256 :param schema_response: The response class whose fields decide the 

257 available filter parameters. A view passes its ``schema_response``, so 

258 a client can filter only on what it can see. 

259 :param model: The SQLAlchemy model the list endpoint queries. Used to verify 

260 each field resolves to a filterable column; non-column fields are 

261 omitted from the generated params. 

262 :param pagination: The view's pagination settings. A ``NoPagination`` or 

263 ``None`` is an unpaginated view, which returns the full result set. 

264 """ 

265 fields: dict[str, Any] = {} 

266 if isinstance(pagination, NumberedPagination): 

267 page_description = "1-based page number." 

268 if pagination.max_page is not None: 

269 page_description += f" At most {pagination.max_page}." 

270 fields[pagination.page_query_param] = ( 

271 Annotated[ 

272 int, Field(ge=1, le=pagination.max_page, description=page_description) 

273 ], 

274 1, 

275 ) 

276 fields[pagination.page_size_query_param] = ( 

277 Annotated[ 

278 Optional[int], 

279 Field( 

280 ge=1, 

281 le=pagination.max_page_size, 

282 description=( 

283 f"Number of items per page (1–{pagination.max_page_size}). " 

284 f"Defaults to {pagination.default_page_size}." 

285 ), 

286 ), 

287 ], 

288 pagination.default_page_size, 

289 ) 

290 fields[_SORT_QUERY_PARAM] = ( 

291 Annotated[ 

292 Optional[str], 

293 Field( 

294 description=( 

295 "Comma-separated list of fields to sort by. Prefix a " 

296 "field with ``-`` for descending order. Example: " 

297 "``-created_at,name``." 

298 ) 

299 ), 

300 ], 

301 None, 

302 ) 

303 reserved = _reserved_names(pagination) 

304 for name, field in _iter_fields_including_nested(schema_response): 

305 if name in reserved: 

306 raise ValueError( 

307 f"List params for {_written_name(schema_response)!r} cannot expose " 

308 f"field {name!r}: it collides with a reserved pagination/sort " 

309 "parameter. Add a Pydantic alias to expose it as a filter, or " 

310 "rename the pagination parameter." 

311 ) 

312 

313 # Only emit params for fields that resolve to a filterable column on the 

314 # model -- using the very predicate the request path applies. A 

315 # relationship/collection field (e.g. ``books: list[BookRef]``) or a 

316 # reference traversal that does not resolve would otherwise advertise 

317 # filters in OpenAPI that always 400 at request time. 

318 try: 

319 _resolve_column(model, name, schema_response) 

320 except BadQueryParam: 

321 continue 

322 

323 # Type filter parameters as ``Optional[list[str]]`` instead of the 

324 # column's true type) so FastAPI/Starlette preserve repeated query 

325 # parameters as a list and downstream ``_parse_value`` can perform 

326 # field-type coercion. ``__isnull`` stays a scalar bool because 

327 # repeating it makes no sense. 

328 # Collection/JSON-typed fields (``dict``, ``list[str]``) resolve to 

329 # real columns, but no query-string value can coerce into them, so 

330 # every operator except ``__isnull`` would 400 at request time. 

331 supports_scalar = _supports_scalar_operators(field) 

332 if supports_scalar: 

333 eq_desc = ( 

334 f"Filter by ``{name}``. Comma-separated values are OR-combined " 

335 "(SQL ``IN``). Repeat the parameter to AND multiple predicates." 

336 ) 

337 ne_desc = ( 

338 f"Exclude rows where ``{name}`` matches. Comma-separated values " 

339 "are AND-combined (SQL ``NOT IN``)." 

340 ) 

341 in_desc = ( 

342 f"Filter by ``{name}`` with explicit SQL ``IN`` semantics. " 

343 "Provide comma-separated values." 

344 ) 

345 fields[name] = ( 

346 Annotated[Optional[list[str]], Field(description=eq_desc)], 

347 None, 

348 ) 

349 fields[f"{name}__in"] = ( 

350 Annotated[Optional[list[str]], Field(description=in_desc)], 

351 None, 

352 ) 

353 fields[f"{name}__ne"] = ( 

354 Annotated[Optional[list[str]], Field(description=ne_desc)], 

355 None, 

356 ) 

357 fields[f"{name}__isnull"] = ( 

358 Annotated[ 

359 Optional[bool], 

360 Field( 

361 description=( 

362 f"``true`` matches rows where ``{name}`` IS NULL; " 

363 f"``false`` matches IS NOT NULL." 

364 ) 

365 ), 

366 ], 

367 None, 

368 ) 

369 

370 if supports_scalar and _supports_range_operators(field): 

371 for suffix, sql in ( 

372 ("__gte", ">="), 

373 ("__lte", "<="), 

374 ("__gt", ">"), 

375 ("__lt", "<"), 

376 ): 

377 fields[f"{name}{suffix}"] = ( 

378 Annotated[ 

379 Optional[list[str]], 

380 Field(description=f"``{name} {sql} value``."), 

381 ], 

382 None, 

383 ) 

384 

385 if _is_string_field(field): 

386 fields[f"{name}__contains"] = ( 

387 Annotated[ 

388 Optional[list[str]], 

389 Field( 

390 description=( 

391 f"Case-sensitive substring search on " 

392 f"``{name}``. Repeat the parameter to AND " 

393 "multiple terms; whitespace inside one value is " 

394 "also AND-split as a convenience." 

395 ) 

396 ), 

397 ], 

398 None, 

399 ) 

400 fields[f"{name}__icontains"] = ( 

401 Annotated[ 

402 Optional[list[str]], 

403 Field( 

404 description=( 

405 f"Case-insensitive substring search on " 

406 f"``{name}``. Repeat the parameter to AND " 

407 "multiple terms; whitespace inside one value is " 

408 "also AND-split as a convenience." 

409 ) 

410 ), 

411 ], 

412 None, 

413 ) 

414 

415 schema_name = _schema_role_name(schema_response, "ListParams") 

416 validators = ( 

417 {"_validate_pagination_offset": _pagination_offset_validator(pagination)} 

418 if isinstance(pagination, NumberedPagination) 

419 else {} 

420 ) 

421 params_model = pydantic.create_model( # type: ignore[call-overload] 

422 schema_name, __validators__=validators, **fields 

423 ) 

424 _CREATED_WITH[params_model] = pagination 

425 return params_model 

426 

427 

428def apply_list_params( 

429 query: Select[Any], 

430 list_params: pydantic.BaseModel | QueryParams, 

431 model: type[DeclarativeBase], 

432 schema_response: SchemaType, 

433 *, 

434 pagination: _AnyPagination = cast(Any, _FROM_PARAMS), 

435) -> Select[Any]: 

436 """ 

437 Apply pagination, sorting, and filtering to ``query`` using validated 

438 list params. 

439 

440 ``list_params`` is normally an instance of the model returned by 

441 :func:`derive_schema_list_params`. The default list endpoints always pass a 

442 validated instance, so pagination/filter bounds have already been checked. 

443 ``schema_response`` decides the fields a client can filter and sort on. 

444 A view passes its response class. 

445 

446 The arguments match the view method 

447 :meth:`~fastapi_restly.views.RestView.apply_list_params`, which calls this 

448 function with the view's model, response class and pagination. 

449 

450 A raw :class:`~starlette.datastructures.QueryParams` is also accepted 

451 for callers that build the query parameters programmatically. 

452 **Raw inputs bypass schema validation** — the caller is responsible 

453 for verifying ``page``/``page_size`` ranges and any per-view bounds 

454 (``max_page_size``); this function only performs the minimum coercion 

455 needed to apply the SQL clauses. 

456 

457 ``pagination`` names the page and page-size parameters to read. By default 

458 it is the pagination that :func:`derive_schema_list_params` created the 

459 list params model with, and ``page`` and ``page_size`` for raw 

460 ``QueryParams`` or a hand-written model. A ``NoPagination`` or ``None`` 

461 applies no ``LIMIT``/``OFFSET``, and so does a missing page-size key. 

462 

463 Examples:: 

464 

465 # Pagination 

466 page=2&page_size=50 

467 

468 # Sorting 

469 sort=name,-created_at 

470 

471 # Filtering 

472 name=Bob&status=active&created_at__gte=2024-01-01 

473 

474 # Contains (string fields) 

475 name__contains=John&email__icontains=example 

476 """ 

477 if isinstance(pagination, _FromParams): 

478 pagination = ( 

479 _CREATED_WITH.get(type(list_params), _DEFAULT_PAGINATION) 

480 if isinstance(list_params, pydantic.BaseModel) 

481 else _DEFAULT_PAGINATION 

482 ) 

483 query_params = _coerce_to_query_params(list_params) 

484 aliases: _JoinAliases = {} 

485 query = _apply_filtering( 

486 query_params, 

487 query, 

488 model, 

489 schema_response, 

490 aliases=aliases, 

491 pagination=pagination, 

492 ) 

493 query = _apply_sorting(query_params, query, model, schema_response, aliases=aliases) 

494 if isinstance(pagination, NumberedPagination): 

495 query = _apply_pagination(query_params, query, pagination) 

496 return query 

497 

498 

499def _coerce_to_query_params( 

500 list_params: pydantic.BaseModel | QueryParams, 

501) -> QueryParams: 

502 """Normalise validated list params or raw QueryParams to QueryParams. 

503 

504 When a dumped field is a list (e.g. a repeated ``name__contains``), each 

505 element is expanded to its own ``(key, value)`` tuple so that 

506 ``QueryParams.multi_items()`` later returns the original repeated values. 

507 """ 

508 if isinstance(list_params, QueryParams): 

509 return list_params 

510 if isinstance(list_params, pydantic.BaseModel): 

511 dumped = list_params.model_dump(exclude_none=True, by_alias=True, mode="json") 

512 items: list[tuple[str, str]] = [] 

513 for key, value in dumped.items(): 

514 if isinstance(value, list): 

515 items.extend((key, str(item)) for item in value) 

516 else: 

517 items.append((key, str(value))) 

518 return QueryParams(items) 

519 return QueryParams(list_params) 

520 

521 

522def _apply_pagination( 

523 query_params: QueryParams, select_query: Select[Any], pagination: NumberedPagination 

524) -> Select[Any]: 

525 page_size = _get_int(query_params, pagination.page_size_query_param) 

526 if page_size is None: 

527 return select_query 

528 page = _get_int(query_params, pagination.page_query_param) or 1 

529 offset = (page - 1) * page_size 

530 return select_query.limit(page_size).offset(offset) 

531 

532 

533def _get_int(query_params: QueryParams, param_name: str) -> Optional[int]: 

534 value = query_params.get(param_name) 

535 if not value: 

536 return None 

537 try: 

538 return int(value) 

539 except ValueError: 

540 raise BadQueryParam( 

541 f"Invalid value for URL query parameter {param_name}: " 

542 f"{value} is not an integer" 

543 ) 

544 

545 

546def _apply_sorting( 

547 query_params: QueryParams, 

548 select_query: Select[Any], 

549 model: type[DeclarativeBase], 

550 schema_cls: SchemaType, 

551 *, 

552 aliases: _JoinAliases | None = None, 

553) -> Select[Any]: 

554 if aliases is None: 

555 aliases = {} 

556 sorted_on: list[InstrumentedAttribute[Any]] = [] 

557 for column_name in (query_params.get("sort") or "").split(","): 

558 if not column_name: 

559 continue 

560 order = sqlalchemy.asc 

561 if column_name.startswith("-"): 

562 order = sqlalchemy.desc 

563 column_name = column_name[1:] 

564 select_query, column = _join_column( 

565 select_query, model, column_name, schema_cls, aliases 

566 ) 

567 select_query = select_query.order_by(order(column)) 

568 sorted_on.append(column) 

569 return _append_pk_tiebreak(select_query, model, sorted_on) 

570 

571 

572def _written_name(schema_cls: SchemaType) -> str: 

573 """The name of the class that the user wrote: for a response class that 

574 Restly derived, the view's schema.""" 

575 return schema_cls.__dict__.get("__restly_response_of__", schema_cls).__name__ 

576 

577 

578def _iter_fields_including_nested( 

579 schema_cls: SchemaType, prefix: str = "" 

580) -> Iterator[tuple[str, FieldInfo]]: 

581 for name, field in schema_cls.model_fields.items(): 

582 # A WriteOnly field, or one with ``exclude=True``, never goes out, 

583 # so it is never a filter or sort key. 

584 if field.exclude is True: 

585 continue 

586 public_name = field.alias or name 

587 # Each segment of the public dotted path becomes part of the URL 

588 # grammar. ``__`` is reserved for operator suffixes (``__gte``, 

589 # ``__contains``, ...) and ``.`` is reserved for relation traversal, 

590 # so a segment containing either character would create an 

591 # ambiguous URL key. Reject at schema-generation time so the 

592 # collision surfaces during view registration, not at request time. 

593 if "__" in public_name: 

594 raise ValueError( 

595 f"List params for {_written_name(schema_cls)!r} cannot " 

596 f"expose field {public_name!r}: ``__`` is reserved for " 

597 "operator suffixes. Choose a different Pydantic alias." 

598 ) 

599 if "." in public_name: 

600 raise ValueError( 

601 f"List params for {_written_name(schema_cls)!r} cannot " 

602 f"expose field {public_name!r}: ``.`` is reserved for " 

603 "relation traversal. Choose a different Pydantic alias." 

604 ) 

605 full_name = f"{prefix}.{public_name}" if prefix else public_name 

606 nested = _get_nested_schema(field) 

607 if nested and not _is_idref_field(field): 

608 yield from _iter_fields_including_nested(nested, full_name) 

609 else: 

610 yield full_name, field 

611 

612 

613def _resolve_field_name(schema_cls: SchemaType, public_name: str) -> str | None: 

614 """Return the Python field name for a public URL field name. 

615 

616 The public name is the field's alias when one is declared, otherwise the 

617 field name itself. Aliased fields are *only* reachable by their alias — 

618 Python field names are never part of the public URL contract, even when 

619 the schema has ``populate_by_name=True`` (which only affects how Pydantic 

620 parses input bodies, not the generated list params). 

621 

622 A field that never goes out, WriteOnly or with ``exclude=True``, never 

623 resolves: filtering or sorting on it would let a client read back a value 

624 that responses leave out. 

625 """ 

626 resolved: str | None = None 

627 for field_name, field in schema_cls.model_fields.items(): 

628 if field.alias == public_name: 

629 resolved = field_name 

630 break 

631 else: 

632 field = schema_cls.model_fields.get(public_name) 

633 if field is not None and field.alias is None: 

634 resolved = public_name 

635 if resolved is not None and schema_cls.model_fields[resolved].exclude is True: 

636 return None 

637 return resolved 

638 

639 

640def _join_column( 

641 select_query: Select[Any], 

642 model: type[DeclarativeBase], 

643 column_path: str, 

644 schema_cls: SchemaType, 

645 aliases: _JoinAliases, 

646) -> tuple[Select[Any], InstrumentedAttribute[Any]]: 

647 """Join each relationship path once and return its aliased leaf column.""" 

648 relationships, column = _resolve_column(model, column_path, schema_cls) 

649 source: type[DeclarativeBase] | AliasedClass[Any] = model 

650 path: tuple[str, ...] = () 

651 for relationship in relationships: 

652 # The full path distinguishes home_city.country from work_city.country 

653 # and successive hops through a self-referential relationship. 

654 path += (relationship.key,) 

655 target = aliases.get(path) 

656 if target is None: 

657 target = aliased(relationship.property.mapper) 

658 select_query = select_query.join_from( 

659 source, getattr(source, relationship.key).of_type(target) 

660 ) 

661 aliases[path] = target 

662 source = target 

663 return select_query, getattr(source, column.key) 

664 

665 

666def _resolve_column( 

667 model: type[DeclarativeBase], column_path: str, schema_cls: SchemaType 

668) -> tuple[list[InstrumentedAttribute[Any]], InstrumentedAttribute[Any]]: 

669 """Resolve a (possibly dotted) public column path to its SQLAlchemy column, 

670 plus the relationship attributes that need to be joined. 

671 

672 Strict: every path segment must resolve through the schema's public name 

673 (alias when set, Python field name otherwise). Falling back to a raw 

674 model attribute lookup would let URLs reach columns the schema didn't 

675 expose — for example, a Python field name on an aliased schema field — 

676 and silently bypass the public-name contract. 

677 """ 

678 joins: list[InstrumentedAttribute[Any]] = [] 

679 current_model = model 

680 current_schema: SchemaType | None = schema_cls 

681 name = column_path 

682 while "." in name: 

683 relation, _, name = name.partition(".") 

684 if current_schema is None: 684 ↛ 685line 684 didn't jump to line 685 because the condition on line 684 was never true

685 raise BadQueryParam(f"Invalid attribute in URL query: {column_path}") 

686 field_name = _resolve_field_name(current_schema, relation) 

687 if field_name is None: 

688 raise BadQueryParam(f"Invalid attribute in URL query: {column_path}") 

689 rel = getattr(current_model, field_name, None) 

690 if not isinstance(rel, InstrumentedAttribute) or not hasattr( 

691 rel.property, "mapper" 

692 ): 

693 raise BadQueryParam(f"Invalid attribute in URL query: {column_path}") 

694 joins.append(rel) 

695 current_model = rel.property.mapper.class_ 

696 current_schema = _get_nested_schema(current_schema.model_fields[field_name]) 

697 

698 if current_schema is None: 698 ↛ 699line 698 didn't jump to line 699 because the condition on line 698 was never true

699 raise BadQueryParam(f"Invalid attribute in URL query: {column_path}") 

700 field_name = _resolve_field_name(current_schema, name) 

701 if field_name is None: 

702 raise BadQueryParam(f"Invalid attribute in URL query: {column_path}") 

703 column = getattr(current_model, field_name, None) 

704 if ( 

705 column is None 

706 or not isinstance(column, InstrumentedAttribute) 

707 or not isinstance(column.property, ColumnProperty) 

708 ): 

709 raise BadQueryParam(f"Invalid attribute in URL query: {column_path}") 

710 return joins, cast(InstrumentedAttribute[Any], column) 

711 

712 

713def _apply_filtering( 

714 query_params: QueryParams, 

715 select_query: Select[Any], 

716 model: type[DeclarativeBase], 

717 schema_cls: SchemaType, 

718 *, 

719 aliases: _JoinAliases | None = None, 

720 pagination: _AnyPagination = _DEFAULT_PAGINATION, 

721) -> Select[Any]: 

722 """Apply ``key=value`` and ``key__op=value`` filters to ``select_query``. 

723 

724 Multiple filters on the same column are AND-combined. Comma-separated 

725 values within one parameter are OR-combined for ``eq`` (the default), 

726 mapped to SQL ``IN`` for ``in``, and AND-combined for ``ne`` (so 

727 ``status__ne=a,b`` means NOT IN (a, b)). For ``contains``/``icontains`` 

728 values are split on whitespace and AND-combined. 

729 """ 

730 if aliases is None: 

731 aliases = {} 

732 reserved = _reserved_names(pagination) 

733 filters: dict[InstrumentedAttribute[Any], list[ColumnElement[Any]]] = defaultdict( 

734 list 

735 ) 

736 

737 for key, raw_value in query_params.multi_items(): 

738 if key in reserved: 

739 continue 

740 

741 if "__" in key: 

742 column_name, op = key.split("__", 1) 

743 else: 

744 column_name, op = key, "eq" 

745 

746 select_query, column = _join_column( 

747 select_query, model, column_name, schema_cls, aliases 

748 ) 

749 parser = functools.partial(_parse_value, schema_cls, column_name) 

750 

751 if op == "isnull": 

752 try: 

753 value = pydantic.TypeAdapter(bool).validate_python(raw_value) 

754 except pydantic.ValidationError as exc: 

755 raise BadQueryParam( 

756 f"Invalid value for URL query parameter {key}" 

757 ) from exc 

758 filters[column].append(column.is_(None) if value else column.isnot(None)) 

759 continue 

760 

761 clause = _build_clause(column, raw_value, op, parser) 

762 if clause is not None: 

763 filters[column].append(clause) 

764 

765 for column, clauses in filters.items(): 

766 and_clause = clauses[0] if len(clauses) == 1 else sqlalchemy.and_(*clauses) 

767 select_query = select_query.where(and_clause) 

768 return select_query 

769 

770 

771def _build_clause( 

772 column: InstrumentedAttribute[Any], 

773 raw_value: str, 

774 op: str, 

775 parser: Callable[[str], Any], 

776) -> ColumnElement[Any] | None: 

777 """Combine multiple values within one parameter according to ``op`` semantics.""" 

778 if op in {"contains", "icontains"}: 

779 values = [v for v in raw_value.split() if v] 

780 if not values: 

781 return None 

782 clauses = [_make_where_clause(column, v, op, parser) for v in values] 

783 return clauses[0] if len(clauses) == 1 else sqlalchemy.and_(*clauses) 

784 

785 values = raw_value.split(",") 

786 if not values: 786 ↛ 787line 786 didn't jump to line 787 because the condition on line 786 was never true

787 return None 

788 if op == "in": 

789 return column.in_([_parse_filter_value(column, v, parser) for v in values]) 

790 clauses = [_make_where_clause(column, v, op, parser) for v in values] 

791 if len(clauses) == 1: 

792 return clauses[0] 

793 # ``ne`` with multiple values means NOT IN (...) — AND-combine, not OR. 

794 if op == "ne": 

795 return sqlalchemy.and_(*clauses) 

796 return sqlalchemy.or_(*clauses) 

797 

798 

799def _parse_value(schema_cls: SchemaType, column_name: str, value: str) -> Any: 

800 if "." in column_name: 

801 relation, _, column_part = column_name.partition(".") 

802 relation_field_name = _resolve_field_name(schema_cls, relation) or relation 

803 field = schema_cls.model_fields.get(relation_field_name) 

804 nested = _get_nested_schema(field) 

805 if nested is None: 805 ↛ 806line 805 didn't jump to line 806 because the condition on line 805 was never true

806 raise BadQueryParam(f"Invalid attribute in URL query: {column_name}") 

807 return _parse_value(nested, column_part, value) 

808 

809 field_name = _resolve_field_name(schema_cls, column_name) 

810 if field_name is None: 

811 raise BadQueryParam(f"Invalid attribute in URL query: {column_name}") 

812 

813 try: 

814 values, _, _ = _field_validator(schema_cls, field_name).validate_python( 

815 {field_name: value} 

816 ) 

817 result = values[field_name] 

818 except Exception: 

819 raise BadQueryParam(f"Invalid attribute in URL query: {column_name}") 

820 # An IDRef[T] FK field validates to an IDRef object; the SQL bind value 

821 # is its scalar id, not the reference wrapper (which cannot bind). 

822 if isinstance(result, IDSchema): 

823 return result.id 

824 return result 

825 

826 

827@functools.cache 

828def _field_validator(schema_cls: SchemaType, field_name: str) -> SchemaValidator: 

829 """A validator for a one-field mapping, cut from the model's core schema. 

830 

831 The named field supplies ``ValidationInfo.field_name``. No other fields 

832 are validated, so ``ValidationInfo.data`` is empty. 

833 

834 Field validators, constraints and the model config apply; model validators 

835 do not. A filter value is one column, so a cross-field rule has nothing to 

836 compare against and would only reject legal filters (as validating through 

837 ``validate_assignment`` on a ``model_construct()`` skeleton did). 

838 """ 

839 core: Any = schema_cls.__pydantic_core_schema__ 

840 definitions = core.get("definitions") if core.get("type") == "definitions" else None 

841 definitions_by_ref = { 

842 definition["ref"]: definition for definition in definitions or () 

843 } 

844 config = None 

845 node: Any = core 

846 while node is not None and node.get("type") != "model-fields": 

847 # Recursive models keep their model schema in the definitions. 

848 if node.get("type") == "definition-ref": 

849 node = definitions_by_ref[node["schema_ref"]] 

850 continue 

851 if node.get("type") == "model" and config is None: 

852 config = node.get("config") 

853 node = node.get("schema") 

854 if node is None: 854 ↛ 855line 854 didn't jump to line 855 because the condition on line 854 was never true

855 raise LookupError(f"{schema_cls.__name__} has no model-fields core schema") 

856 field_schema: Any = core_schema.model_fields_schema( 

857 {field_name: core_schema.model_field(node["fields"][field_name]["schema"])} 

858 ) 

859 if definitions: