Get a **query** message.
(
options: int,
collection_name: str,
num_to_skip: int,
num_to_return: int,
query: Mapping[str, Any],
field_selector: Optional[Mapping[str, Any]],
opts: CodecOptions[Any],
ctx: Union[SnappyContext, ZlibContext, ZstdContext, None] = None,
)
| 494 | |
| 495 | |
| 496 | def _query( |
| 497 | options: int, |
| 498 | collection_name: str, |
| 499 | num_to_skip: int, |
| 500 | num_to_return: int, |
| 501 | query: Mapping[str, Any], |
| 502 | field_selector: Optional[Mapping[str, Any]], |
| 503 | opts: CodecOptions[Any], |
| 504 | ctx: Union[SnappyContext, ZlibContext, ZstdContext, None] = None, |
| 505 | ) -> tuple[int, bytes, int]: |
| 506 | """Get a **query** message.""" |
| 507 | if ctx: |
| 508 | return _query_compressed( |
| 509 | options, collection_name, num_to_skip, num_to_return, query, field_selector, opts, ctx |
| 510 | ) |
| 511 | return _query_uncompressed( |
| 512 | options, collection_name, num_to_skip, num_to_return, query, field_selector, opts |
| 513 | ) |
| 514 | |
| 515 | |
| 516 | _pack_long_long = struct.Struct("<q").pack |
no test coverage detected