Get the spec to use for a query.
(self)
| 305 | return self.__class__(self._collection, session=session) |
| 306 | |
| 307 | def _query_spec(self) -> Mapping[str, Any]: |
| 308 | """Get the spec to use for a query.""" |
| 309 | operators: dict[str, Any] = {} |
| 310 | if self._ordering: |
| 311 | operators["$orderby"] = self._ordering |
| 312 | if self._explain: |
| 313 | operators["$explain"] = True |
| 314 | if self._hint: |
| 315 | operators["$hint"] = self._hint |
| 316 | if self._let: |
| 317 | operators["let"] = self._let |
| 318 | if self._comment: |
| 319 | operators["$comment"] = self._comment |
| 320 | if self._max_scan: |
| 321 | operators["$maxScan"] = self._max_scan |
| 322 | if self._max_time_ms is not None: |
| 323 | operators["$maxTimeMS"] = self._max_time_ms |
| 324 | if self._max: |
| 325 | operators["$max"] = self._max |
| 326 | if self._min: |
| 327 | operators["$min"] = self._min |
| 328 | if self._return_key is not None: |
| 329 | operators["$returnKey"] = self._return_key |
| 330 | if self._show_record_id is not None: |
| 331 | # This is upgraded to showRecordId for MongoDB 3.2+ "find" command. |
| 332 | operators["$showDiskLoc"] = self._show_record_id |
| 333 | if self._snapshot is not None: |
| 334 | operators["$snapshot"] = self._snapshot |
| 335 | |
| 336 | if operators: |
| 337 | # Make a shallow copy so we can cleanly rewind or clone. |
| 338 | spec = dict(self._spec) |
| 339 | |
| 340 | # Allow-listed commands must be wrapped in $query. |
| 341 | if "$query" not in spec: |
| 342 | # $query has to come first |
| 343 | spec = {"$query": spec} |
| 344 | |
| 345 | spec.update(operators) |
| 346 | return spec |
| 347 | # Have to wrap with $query if "query" is the first key. |
| 348 | # We can't just use $query anytime "query" is a key as |
| 349 | # that breaks commands like count and find_and_modify. |
| 350 | # Checking spec.keys()[0] covers the case that the spec |
| 351 | # was passed as an instance of SON or OrderedDict. |
| 352 | elif "query" in self._spec and (len(self._spec) == 1 or next(iter(self._spec)) == "query"): |
| 353 | return {"$query": self._spec} |
| 354 | |
| 355 | return self._spec |
| 356 | |
| 357 | def _check_okay_to_chain(self) -> None: |
| 358 | """Check if it is okay to chain more options onto this cursor.""" |