codekingpro/portable-devtools
114k
1# engine/result.py
2# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7
8"""Define generic result set constructs."""
9
10from __future__ import annotations
11
12from enum import Enum
13import functools
14import itertools
15import operator
16import typing
17from typing import Any
18from typing import Callable
19from typing import cast
20from typing import Dict
21from typing import Generic
22from typing import Iterable
23from typing import Iterator
24from typing import List
25from typing import Mapping
26from typing import NoReturn
27from typing import Optional
28from typing import overload
29from typing import Sequence
30from typing import Set
31from typing import Tuple
32from typing import TYPE_CHECKING
33from typing import TypeVar
34from typing import Union
35
36from .row import Row
37from .row import RowMapping
38from .. import exc
39from .. import util
40from ..sql.base import _generative
41from ..sql.base import HasMemoized
42from ..sql.base import InPlaceGenerative
43from ..util import HasMemoized_ro_memoized_attribute
44from ..util import NONE_SET
45from ..util._has_cy import HAS_CYEXTENSION
46from ..util.typing import Literal
47from ..util.typing import Self
48
49if typing.TYPE_CHECKING or not HAS_CYEXTENSION:
50 from ._py_row import tuplegetter as tuplegetter
51else:
52 from sqlalchemy.cyextension.resultproxy import tuplegetter as tuplegetter
53
54if typing.TYPE_CHECKING:
55 from ..sql.schema import Column
56 from ..sql.type_api import _ResultProcessorType
57
58_KeyType = Union[str, "Column[Any]"]
59_KeyIndexType = Union[str, "Column[Any]", int]
60
61# is overridden in cursor using _CursorKeyMapRecType
62_KeyMapRecType = Any
63
64_KeyMapType = Mapping[_KeyType, _KeyMapRecType]
65
66
67_RowData = Union[Row[Any], RowMapping, Any]
68"""A generic form of "row" that accommodates for the different kinds of
69"rows" that different result objects return, including row, row mapping, and
70scalar values"""
71
72_RawRowType = Tuple[Any, ...]
73"""represents the kind of row we get from a DBAPI cursor"""
74
75_R = TypeVar("_R", bound=_RowData)
76_T = TypeVar("_T", bound=Any)
77_TP = TypeVar("_TP", bound=Tuple[Any, ...])
78
79_InterimRowType = Union[_R, _RawRowType]
80"""a catchall "anything" kind of return type that can be applied
81across all the result types
82
83"""
84
85_InterimSupportsScalarsRowType = Union[Row[Any], Any]
86
87_ProcessorsType = Sequence[Optional["_ResultProcessorType[Any]"]]
88_TupleGetterType = Callable[[Sequence[Any]], Sequence[Any]]
89_UniqueFilterType = Callable[[Any], Any]
90_UniqueFilterStateType = Tuple[Set[Any], Optional[_UniqueFilterType]]
91
92
93class ResultMetaData:
94 """Base for metadata about result rows."""
95
96 __slots__ = ()
97
98 _tuplefilter: Optional[_TupleGetterType] = None
99 _translated_indexes: Optional[Sequence[int]] = None
100 _unique_filters: Optional[Sequence[Callable[[Any], Any]]] = None
101 _keymap: _KeyMapType
102 _keys: Sequence[str]
103 _processors: Optional[_ProcessorsType]
104 _key_to_index: Mapping[_KeyType, int]
105
106 @property
107 def keys(self) -> RMKeyView:
108 return RMKeyView(self)
109
110 def _has_key(self, key: object) -> bool:
111 raise NotImplementedError()
112
113 def _for_freeze(self) -> ResultMetaData:
114 raise NotImplementedError()
115
116 @overload
117 def _key_fallback(
118 self, key: Any, err: Optional[Exception], raiseerr: Literal[True] = ...
119 ) -> NoReturn: ...
120
121 @overload
122 def _key_fallback(
123 self,
124 key: Any,
125 err: Optional[Exception],
126 raiseerr: Literal[False] = ...,
127 ) -> None: ...
128
129 @overload
130 def _key_fallback(
131 self, key: Any, err: Optional[Exception], raiseerr: bool = ...
132 ) -> Optional[NoReturn]: ...
133
134 def _key_fallback(
135 self, key: Any, err: Optional[Exception], raiseerr: bool = True
136 ) -> Optional[NoReturn]:
137 assert raiseerr
138 raise KeyError(key) from err
139
140 def _raise_for_ambiguous_column_name(
141 self, rec: _KeyMapRecType
142 ) -> NoReturn:
143 raise NotImplementedError(
144 "ambiguous column name logic is implemented for "
145 "CursorResultMetaData"
146 )
147
148 def _index_for_key(
149 self, key: _KeyIndexType, raiseerr: bool
150 ) -> Optional[int]:
151 raise NotImplementedError()
152
153 def _indexes_for_keys(
154 self, keys: Sequence[_KeyIndexType]
155 ) -> Sequence[int]:
156 raise NotImplementedError()
157
158 def _metadata_for_keys(
159 self, keys: Sequence[_KeyIndexType]
160 ) -> Iterator[_KeyMapRecType]:
161 raise NotImplementedError()
162
163 def _reduce(self, keys: Sequence[_KeyIndexType]) -> ResultMetaData:
164 raise NotImplementedError()
165
166 def _getter(
167 self, key: Any, raiseerr: bool = True
168 ) -> Optional[Callable[[Row[Any]], Any]]:
169 index = self._index_for_key(key, raiseerr)
170
171 if index is not None:
172 return operator.itemgetter(index)
173 else:
174 return None
175
176 def _row_as_tuple_getter(
177 self, keys: Sequence[_KeyIndexType]
178 ) -> _TupleGetterType:
179 indexes = self._indexes_for_keys(keys)
180 return tuplegetter(*indexes)
181
182 def _make_key_to_index(
183 self, keymap: Mapping[_KeyType, Sequence[Any]], index: int
184 ) -> Mapping[_KeyType, int]:
185 return {
186 key: rec[index]
187 for key, rec in keymap.items()
188 if rec[index] is not None
189 }
190
191 def _key_not_found(self, key: Any, attr_error: bool) -> NoReturn:
192 if key in self._keymap:
193 # the index must be none in this case
194 self._raise_for_ambiguous_column_name(self._keymap[key])
195 else:
196 # unknown key
197 if attr_error:
198 try:
199 self._key_fallback(key, None)
200 except KeyError as ke:
201 raise AttributeError(ke.args[0]) from ke
202 else:
203 self._key_fallback(key, None)
204
205 @property
206 def _effective_processors(self) -> Optional[_ProcessorsType]:
207 if not self._processors or NONE_SET.issuperset(self._processors):
208 return None
209 else:
210 return self._processors
211
212
213class RMKeyView(typing.KeysView[Any]):
214 __slots__ = ("_parent", "_keys")
215
216 _parent: ResultMetaData
217 _keys: Sequence[str]
218
219 def __init__(self, parent: ResultMetaData):
220 self._parent = parent
221 self._keys = [k for k in parent._keys if k is not None]
222
223 def __len__(self) -> int:
224 return len(self._keys)
225
226 def __repr__(self) -> str:
227 return "{0.__class__.__name__}({0._keys!r})".format(self)
228
229 def __iter__(self) -> Iterator[str]:
230 return iter(self._keys)
231
232 def __contains__(self, item: Any) -> bool:
233 if isinstance(item, int):
234 return False
235
236 # note this also includes special key fallback behaviors
237 # which also don't seem to be tested in test_resultset right now
238 return self._parent._has_key(item)
239
240 def __eq__(self, other: Any) -> bool:
241 return list(other) == list(self)
242
243 def __ne__(self, other: Any) -> bool:
244 return list(other) != list(self)
245
246
247class SimpleResultMetaData(ResultMetaData):
248 """result metadata for in-memory collections."""
249
250 __slots__ = (
251 "_keys",
252 "_keymap",
253 "_processors",
254 "_tuplefilter",
255 "_translated_indexes",
256 "_unique_filters",
257 "_key_to_index",
258 )
259
260 _keys: Sequence[str]
261
262 def __init__(
263 self,
264 keys: Sequence[str],
265 extra: Optional[Sequence[Any]] = None,
266 _processors: Optional[_ProcessorsType] = None,
267 _tuplefilter: Optional[_TupleGetterType] = None,
268 _translated_indexes: Optional[Sequence[int]] = None,
269 _unique_filters: Optional[Sequence[Callable[[Any], Any]]] = None,
270 ):
271 self._keys = list(keys)
272 self._tuplefilter = _tuplefilter
273 self._translated_indexes = _translated_indexes
274 self._unique_filters = _unique_filters
275 if extra:
276 recs_names = [
277 (
278 (name,) + (extras if extras else ()),
279 (index, name, extras),
280 )
281 for index, (name, extras) in enumerate(zip(self._keys, extra))
282 ]
283 else:
284 recs_names = [
285 ((name,), (index, name, ()))
286 for index, name in enumerate(self._keys)
287 ]
288
289 self._keymap = {key: rec for keys, rec in recs_names for key in keys}
290
291 self._processors = _processors
292
293 self._key_to_index = self._make_key_to_index(self._keymap, 0)
294
295 def _has_key(self, key: object) -> bool:
296 return key in self._keymap
297
298 def _for_freeze(self) -> ResultMetaData:
299 unique_filters = self._unique_filters
300 if unique_filters and self._tuplefilter:
301 unique_filters = self._tuplefilter(unique_filters)
302
303 # TODO: are we freezing the result with or without uniqueness
304 # applied?
305 return SimpleResultMetaData(
306 self._keys,
307 extra=[self._keymap[key][2] for key in self._keys],
308 _unique_filters=unique_filters,
309 )
310
311 def __getstate__(self) -> Dict[str, Any]:
312 return {
313 "_keys": self._keys,
314 "_translated_indexes": self._translated_indexes,
315 }
316
317 def __setstate__(self, state: Dict[str, Any]) -> None:
318 if state["_translated_indexes"]:
319 _translated_indexes = state["_translated_indexes"]
320 _tuplefilter = tuplegetter(*_translated_indexes)
321 else:
322 _translated_indexes = _tuplefilter = None
323 self.__init__( # type: ignore
324 state["_keys"],
325 _translated_indexes=_translated_indexes,
326 _tuplefilter=_tuplefilter,
327 )
328
329 def _index_for_key(self, key: Any, raiseerr: bool = True) -> int:
330 if int in key.__class__.__mro__:
331 key = self._keys[key]
332 try:
333 rec = self._keymap[key]
334 except KeyError as ke:
335 rec = self._key_fallback(key, ke, raiseerr)
336
337 return rec[0] # type: ignore[no-any-return]
338
339 def _indexes_for_keys(self, keys: Sequence[Any]) -> Sequence[int]:
340 return [self._keymap[key][0] for key in keys]
341
342 def _metadata_for_keys(
343 self, keys: Sequence[Any]
344 ) -> Iterator[_KeyMapRecType]:
345 for key in keys:
346 if int in key.__class__.__mro__:
347 key = self._keys[key]
348
349 try:
350 rec = self._keymap[key]
351 except KeyError as ke:
352 rec = self._key_fallback(key, ke, True)
353
354 yield rec
355
356 def _reduce(self, keys: Sequence[Any]) -> ResultMetaData:
357 try:
358 metadata_for_keys = [
359 self._keymap[
360 self._keys[key] if int in key.__class__.__mro__ else key
361 ]
362 for key in keys
363 ]
364 except KeyError as ke:
365 self._key_fallback(ke.args[0], ke, True)
366
367 indexes: Sequence[int]
368 new_keys: Sequence[str]
369 extra: Sequence[Any]
370 indexes, new_keys, extra = zip(*metadata_for_keys)
371
372 if self._translated_indexes:
373 indexes = [self._translated_indexes[idx] for idx in indexes]
374
375 tup = tuplegetter(*indexes)
376
377 new_metadata = SimpleResultMetaData(
378 new_keys,
379 extra=extra,
380 _tuplefilter=tup,
381 _translated_indexes=indexes,
382 _processors=self._processors,
383 _unique_filters=self._unique_filters,
384 )
385
386 return new_metadata
387
388
389def result_tuple(
390 fields: Sequence[str], extra: Optional[Any] = None
391) -> Callable[[Iterable[Any]], Row[Any]]:
392 parent = SimpleResultMetaData(fields, extra)
393 return functools.partial(
394 Row, parent, parent._effective_processors, parent._key_to_index
395 )
396
397
398# a symbol that indicates to internal Result methods that
399# "no row is returned". We can't use None for those cases where a scalar
400# filter is applied to rows.
401class _NoRow(Enum):
402 _NO_ROW = 0
403
404
405_NO_ROW = _NoRow._NO_ROW
406
407
408class ResultInternal(InPlaceGenerative, Generic[_R]):
409 __slots__ = ()
410
411 _real_result: Optional[Result[Any]] = None
412 _generate_rows: bool = True
413 _row_logging_fn: Optional[Callable[[Any], Any]]
414
415 _unique_filter_state: Optional[_UniqueFilterStateType] = None
416 _post_creational_filter: Optional[Callable[[Any], Any]] = None
417 _is_cursor = False
418
419 _metadata: ResultMetaData
420
421 _source_supports_scalars: bool
422
423 def _fetchiter_impl(self) -> Iterator[_InterimRowType[Row[Any]]]:
424 raise NotImplementedError()
425
426 def _fetchone_impl(
427 self, hard_close: bool = False
428 ) -> Optional[_InterimRowType[Row[Any]]]:
429 raise NotImplementedError()
430
431 def _fetchmany_impl(
432 self, size: Optional[int] = None
433 ) -> List[_InterimRowType[Row[Any]]]:
434 raise NotImplementedError()
435
436 def _fetchall_impl(self) -> List[_InterimRowType[Row[Any]]]:
437 raise NotImplementedError()
438
439 def _soft_close(self, hard: bool = False) -> None:
440 raise NotImplementedError()
441
442 @HasMemoized_ro_memoized_attribute
443 def _row_getter(self) -> Optional[Callable[..., _R]]:
444 real_result: Result[Any] = (
445 self._real_result
446 if self._real_result
447 else cast("Result[Any]", self)
448 )
449
450 if real_result._source_supports_scalars:
451 if not self._generate_rows:
452 return None
453 else:
454 _proc = Row
455
456 def process_row(
457 metadata: ResultMetaData,
458 processors: Optional[_ProcessorsType],
459 key_to_index: Mapping[_KeyType, int],
460 scalar_obj: Any,
461 ) -> Row[Any]:
462 return _proc(
463 metadata, processors, key_to_index, (scalar_obj,)
464 )
465
466 else:
467 process_row = Row # type: ignore
468
469 metadata = self._metadata
470
471 key_to_index = metadata._key_to_index
472 processors = metadata._effective_processors
473 tf = metadata._tuplefilter
474
475 if tf and not real_result._source_supports_scalars:
476 if processors:
477 processors = tf(processors)
478
479 _make_row_orig: Callable[..., _R] = functools.partial( # type: ignore # noqa E501
480 process_row, metadata, processors, key_to_index
481 )
482
483 fixed_tf = tf
484
485 def make_row(row: _InterimRowType[Row[Any]]) -> _R:
486 return _make_row_orig(fixed_tf(row))
487
488 else:
489 make_row = functools.partial( # type: ignore
490 process_row, metadata, processors, key_to_index
491 )
492
493 if real_result._row_logging_fn:
494 _log_row = real_result._row_logging_fn
495 _make_row = make_row
496
497 def make_row(row: _InterimRowType[Row[Any]]) -> _R:
498 return _log_row(_make_row(row)) # type: ignore
499
500 return make_row
501
502 @HasMemoized_ro_memoized_attribute
503 def _iterator_getter(self) -> Callable[..., Iterator[_R]]:
504 make_row = self._row_getter
505
506 post_creational_filter = self._post_creational_filter
507
508 if self._unique_filter_state:
509 uniques, strategy = self._unique_strategy
510
511 def iterrows(self: Result[Any]) -> Iterator[_R]:
512 for raw_row in self._fetchiter_impl():
513 obj: _InterimRowType[Any] = (
514 make_row(raw_row) if make_row else raw_row
515 )
516 hashed = strategy(obj) if strategy else obj
517 if hashed in uniques:
518 continue
519 uniques.add(hashed)
520 if post_creational_filter:
521 obj = post_creational_filter(obj)
522 yield obj # type: ignore
523
524 else:
525
526 def iterrows(self: Result[Any]) -> Iterator[_R]:
527 for raw_row in self._fetchiter_impl():
528 row: _InterimRowType[Any] = (
529 make_row(raw_row) if make_row else raw_row
530 )
531 if post_creational_filter:
532 row = post_creational_filter(row)
533 yield row # type: ignore
534
535 return iterrows
536
537 def _raw_all_rows(self) -> List[_R]:
538 make_row = self._row_getter
539 assert make_row is not None
540 rows = self._fetchall_impl()
541 return [make_row(row) for row in rows]
542
543 def _allrows(self) -> List[_R]:
544 post_creational_filter = self._post_creational_filter
545
546 make_row = self._row_getter
547
548 rows = self._fetchall_impl()
549 made_rows: List[_InterimRowType[_R]]
550 if make_row:
551 made_rows = [make_row(row) for row in rows]
552 else:
553 made_rows = rows # type: ignore
554
555 interim_rows: List[_R]
556
557 if self._unique_filter_state:
558 uniques, strategy = self._unique_strategy
559
560 interim_rows = [
561 made_row # type: ignore
562 for made_row, sig_row in [
563 (
564 made_row,
565 strategy(made_row) if strategy else made_row,
566 )
567 for made_row in made_rows
568 ]
569 if sig_row not in uniques and not uniques.add(sig_row) # type: ignore # noqa: E501
570 ]
571 else:
572 interim_rows = made_rows # type: ignore
573
574 if post_creational_filter:
575 interim_rows = [
576 post_creational_filter(row) for row in interim_rows
577 ]
578 return interim_rows
579
580 @HasMemoized_ro_memoized_attribute
581 def _onerow_getter(
582 self,
583 ) -> Callable[..., Union[Literal[_NoRow._NO_ROW], _R]]:
584 make_row = self._row_getter
585
586 post_creational_filter = self._post_creational_filter
587
588 if self._unique_filter_state:
589 uniques, strategy = self._unique_strategy
590
591 def onerow(self: Result[Any]) -> Union[_NoRow, _R]:
592 _onerow = self._fetchone_impl
593 while True:
594 row = _onerow()
595 if row is None:
596 return _NO_ROW
597 else:
598 obj: _InterimRowType[Any] = (
599 make_row(row) if make_row else row
600 )
601 hashed = strategy(obj) if strategy else obj
602 if hashed in uniques:
603 continue
604 else:
605 uniques.add(hashed)
606 if post_creational_filter:
607 obj = post_creational_filter(obj)
608 return obj # type: ignore
609
610 else:
611
612 def onerow(self: Result[Any]) -> Union[_NoRow, _R]:
613 row = self._fetchone_impl()
614 if row is None:
615 return _NO_ROW
616 else:
617 interim_row: _InterimRowType[Any] = (
618 make_row(row) if make_row else row
619 )
620 if post_creational_filter:
621 interim_row = post_creational_filter(interim_row)
622 return interim_row # type: ignore
623
624 return onerow
625
626 @HasMemoized_ro_memoized_attribute
627 def _manyrow_getter(self) -> Callable[..., List[_R]]:
628 make_row = self._row_getter
629
630 post_creational_filter = self._post_creational_filter
631
632 if self._unique_filter_state:
633 uniques, strategy = self._unique_strategy
634
635 def filterrows(
636 make_row: Optional[Callable[..., _R]],
637 rows: List[Any],
638 strategy: Optional[Callable[[List[Any]], Any]],
639 uniques: Set[Any],
640 ) -> List[_R]:
641 if make_row:
642 rows = [make_row(row) for row in rows]
643
644 if strategy:
645 made_rows = (
646 (made_row, strategy(made_row)) for made_row in rows
647 )
648 else:
649 made_rows = ((made_row, made_row) for made_row in rows)
650 return [
651 made_row
652 for made_row, sig_row in made_rows
653 if sig_row not in uniques and not uniques.add(sig_row) # type: ignore # noqa: E501
654 ]
655
656 def manyrows(
657 self: ResultInternal[_R], num: Optional[int]
658 ) -> List[_R]:
659 collect: List[_R] = []
660
661 _manyrows = self._fetchmany_impl
662
663 if num is None:
664 # if None is passed, we don't know the default
665 # manyrows number, DBAPI has this as cursor.arraysize
666 # different DBAPIs / fetch strategies may be different.
667 # do a fetch to find what the number is. if there are
668 # only fewer rows left, then it doesn't matter.
669 real_result = (
670 self._real_result
671 if self._real_result
672 else cast("Result[Any]", self)
673 )
674 if real_result._yield_per:
675 num_required = num = real_result._yield_per
676 else:
677 rows = _manyrows(num)
678 num = len(rows)
679 assert make_row is not None
680 collect.extend(
681 filterrows(make_row, rows, strategy, uniques)
682 )
683 num_required = num - len(collect)
684 else:
685 num_required = num
686
687 assert num is not None
688
689 while num_required:
690 rows = _manyrows(num_required)
691 if not rows:
692 break
693
694 collect.extend(
695 filterrows(make_row, rows, strategy, uniques)
696 )
697 num_required = num - len(collect)
698
699 if post_creational_filter:
700 collect = [post_creational_filter(row) for row in collect]
701 return collect
702
703 else:
704
705 def manyrows(
706 self: ResultInternal[_R], num: Optional[int]
707 ) -> List[_R]:
708 if num is None:
709 real_result = (
710 self._real_result
711 if self._real_result
712 else cast("Result[Any]", self)
713 )
714 num = real_result._yield_per
715
716 rows: List[_InterimRowType[Any]] = self._fetchmany_impl(num)
717 if make_row:
718 rows = [make_row(row) for row in rows]
719 if post_creational_filter:
720 rows = [post_creational_filter(row) for row in rows]
721 return rows # type: ignore
722
723 return manyrows
724
725 @overload
726 def _only_one_row(
727 self,
728 raise_for_second_row: bool,
729 raise_for_none: Literal[True],
730 scalar: bool,
731 ) -> _R: ...
732
733 @overload
734 def _only_one_row(
735 self,
736 raise_for_second_row: bool,
737 raise_for_none: bool,
738 scalar: bool,
739 ) -> Optional[_R]: ...
740
741 def _only_one_row(
742 self,
743 raise_for_second_row: bool,
744 raise_for_none: bool,
745 scalar: bool,
746 ) -> Optional[_R]:
747 onerow = self._fetchone_impl
748
749 row: Optional[_InterimRowType[Any]] = onerow(hard_close=True)
750 if row is None:
751 if raise_for_none:
752 raise exc.NoResultFound(
753 "No row was found when one was required"
754 )
755 else:
756 return None
757
758 if scalar and self._source_supports_scalars:
759 self._generate_rows = False
760 make_row = None
761 else:
762 make_row = self._row_getter
763
764 try:
765 row = make_row(row) if make_row else row
766 except:
767 self._soft_close(hard=True)
768 raise
769
770 if raise_for_second_row:
771 if self._unique_filter_state:
772 # for no second row but uniqueness, need to essentially
773 # consume the entire result :(
774 uniques, strategy = self._unique_strategy
775
776 existing_row_hash = strategy(row) if strategy else row
777
778 while True:
779 next_row: Any = onerow(hard_close=True)
780 if next_row is None:
781 next_row = _NO_ROW
782 break
783
784 try:
785 next_row = make_row(next_row) if make_row else next_row
786
787 if strategy:
788 assert next_row is not _NO_ROW
789 if existing_row_hash == strategy(next_row):
790 continue
791 elif row == next_row:
792 continue
793 # here, we have a row and it's different
794 break
795 except:
796 self._soft_close(hard=True)
797 raise
798 else:
799 next_row = onerow(hard_close=True)
800 if next_row is None:
801 next_row = _NO_ROW
802
803 if next_row is not _NO_ROW:
804 self._soft_close(hard=True)
805 raise exc.MultipleResultsFound(
806 "Multiple rows were found when exactly one was required"
807 if raise_for_none
808 else "Multiple rows were found when one or none "
809 "was required"
810 )
811 else:
812 next_row = _NO_ROW
813 # if we checked for second row then that would have
814 # closed us :)
815 self._soft_close(hard=True)
816
817 if not scalar:
818 post_creational_filter = self._post_creational_filter
819 if post_creational_filter:
820 row = post_creational_filter(row)
821
822 if scalar and make_row:
823 return row[0] # type: ignore
824 else:
825 return row # type: ignore
826
827 def _iter_impl(self) -> Iterator[_R]:
828 return self._iterator_getter(self)
829
830 def _next_impl(self) -> _R:
831 row = self._onerow_getter(self)
832 if row is _NO_ROW:
833 raise StopIteration()
834 else:
835 return row
836
837 @_generative
838 def _column_slices(self, indexes: Sequence[_KeyIndexType]) -> Self:
839 real_result = (
840 self._real_result
841 if self._real_result
842 else cast("Result[Any]", self)
843 )
844
845 if not real_result._source_supports_scalars or len(indexes) != 1:
846 self._metadata = self._metadata._reduce(indexes)
847
848 assert self._generate_rows
849
850 return self
851
852 @HasMemoized.memoized_attribute
853 def _unique_strategy(self) -> _UniqueFilterStateType:
854 assert self._unique_filter_state is not None
855 uniques, strategy = self._unique_filter_state
856
857 real_result = (
858 self._real_result
859 if self._real_result is not None
860 else cast("Result[Any]", self)
861 )
862
863 if not strategy and self._metadata._unique_filters:
864 if (
865 real_result._source_supports_scalars
866 and not self._generate_rows
867 ):
868 strategy = self._metadata._unique_filters[0]
869 else:
870 filters = self._metadata._unique_filters
871 if self._metadata._tuplefilter:
872 filters = self._metadata._tuplefilter(filters)
873
874 strategy = operator.methodcaller("_filter_on_values", filters)
875 return uniques, strategy
876
877
878class _WithKeys:
879 __slots__ = ()
880
881 _metadata: ResultMetaData
882
883 # used mainly to share documentation on the keys method.
884 def keys(self) -> RMKeyView:
885 """Return an iterable view which yields the string keys that would
886 be represented by each :class:`_engine.Row`.
887
888 The keys can represent the labels of the columns returned by a core
889 statement or the names of the orm classes returned by an orm
890 execution.
891
892 The view also can be tested for key containment using the Python
893 ``in`` operator, which will test both for the string keys represented
894 in the view, as well as for alternate keys such as column objects.
895
896 .. versionchanged:: 1.4 a key view object is returned rather than a
897 plain list.
898
899
900 """
901 return self._metadata.keys
902
903
904class Result(_WithKeys, ResultInternal[Row[_TP]]):
905 """Represent a set of database results.
906
907 .. versionadded:: 1.4 The :class:`_engine.Result` object provides a
908 completely updated usage model and calling facade for SQLAlchemy
909 Core and SQLAlchemy ORM. In Core, it forms the basis of the
910 :class:`_engine.CursorResult` object which replaces the previous
911 :class:`_engine.ResultProxy` interface. When using the ORM, a
912 higher level object called :class:`_engine.ChunkedIteratorResult`
913 is normally used.
914
915 .. note:: In SQLAlchemy 1.4 and above, this object is
916 used for ORM results returned by :meth:`_orm.Session.execute`, which can
917 yield instances of ORM mapped objects either individually or within
918 tuple-like rows. Note that the :class:`_engine.Result` object does not
919 deduplicate instances or rows automatically as is the case with the
920 legacy :class:`_orm.Query` object. For in-Python de-duplication of
921 instances or rows, use the :meth:`_engine.Result.unique` modifier
922 method.
923
924 .. seealso::
925
926 :ref:`tutorial_fetching_rows` - in the :doc:`/tutorial/index`
927
928 """
929
930 __slots__ = ("_metadata", "__dict__")
931
932 _row_logging_fn: Optional[Callable[[Row[Any]], Row[Any]]] = None
933
934 _source_supports_scalars: bool = False
935
936 _yield_per: Optional[int] = None
937
938 _attributes: util.immutabledict[Any, Any] = util.immutabledict()
939
940 def __init__(self, cursor_metadata: ResultMetaData):
941 self._metadata = cursor_metadata
942
943 def __enter__(self) -> Self:
944 return self
945
946 def __exit__(self, type_: Any, value: Any, traceback: Any) -> None:
947 self.close()
948
949 def close(self) -> None:
950 """close this :class:`_engine.Result`.
951
952 The behavior of this method is implementation specific, and is
953 not implemented by default. The method should generally end
954 the resources in use by the result object and also cause any
955 subsequent iteration or row fetching to raise
956 :class:`.ResourceClosedError`.
957
958 .. versionadded:: 1.4.27 - ``.close()`` was previously not generally
959 available for all :class:`_engine.Result` classes, instead only
960 being available on the :class:`_engine.CursorResult` returned for
961 Core statement executions. As most other result objects, namely the
962 ones used by the ORM, are proxying a :class:`_engine.CursorResult`
963 in any case, this allows the underlying cursor result to be closed
964 from the outside facade for the case when the ORM query is using
965 the ``yield_per`` execution option where it does not immediately
966 exhaust and autoclose the database cursor.
967
968 """
969 self._soft_close(hard=True)
970
971 @property
972 def _soft_closed(self) -> bool:
973 raise NotImplementedError()
974
975 @property
976 def closed(self) -> bool:
977 """return ``True`` if this :class:`_engine.Result` reports .closed
978
979 .. versionadded:: 1.4.43
980
981 """
982 raise NotImplementedError()
983
984 @_generative
985 def yield_per(self, num: int) -> Self:
986 """Configure the row-fetching strategy to fetch ``num`` rows at a time.
987
988 This impacts the underlying behavior of the result when iterating over
989 the result object, or otherwise making use of methods such as
990 :meth:`_engine.Result.fetchone` that return one row at a time. Data
991 from the underlying cursor or other data source will be buffered up to
992 this many rows in memory, and the buffered collection will then be
993 yielded out one row at a time or as many rows are requested. Each time
994 the buffer clears, it will be refreshed to this many rows or as many
995 rows remain if fewer remain.
996
997 The :meth:`_engine.Result.yield_per` method is generally used in
998 conjunction with the
999 :paramref:`_engine.Connection.execution_options.stream_results`
1000 execution option, which will allow the database dialect in use to make
1001 use of a server side cursor, if the DBAPI supports a specific "server
1002 side cursor" mode separate from its default mode of operation.
1003
1004 .. tip::
1005
1006 Consider using the
1007 :paramref:`_engine.Connection.execution_options.yield_per`
1008 execution option, which will simultaneously set
1009 :paramref:`_engine.Connection.execution_options.stream_results`
1010 to ensure the use of server side cursors, as well as automatically
1011 invoke the :meth:`_engine.Result.yield_per` method to establish
1012 a fixed row buffer size at once.
1013
1014 The :paramref:`_engine.Connection.execution_options.yield_per`
1015 execution option is available for ORM operations, with
1016 :class:`_orm.Session`-oriented use described at
1017 :ref:`orm_queryguide_yield_per`. The Core-only version which works
1018 with :class:`_engine.Connection` is new as of SQLAlchemy 1.4.40.
1019
1020 .. versionadded:: 1.4
1021
1022 :param num: number of rows to fetch each time the buffer is refilled.
1023 If set to a value below 1, fetches all rows for the next buffer.
1024
1025 .. seealso::
1026
1027 :ref:`engine_stream_results` - describes Core behavior for
1028 :meth:`_engine.Result.yield_per`
1029
1030 :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`
1031
1032 """
1033 self._yield_per = num
1034 return self
1035
1036 @_generative
1037 def unique(self, strategy: Optional[_UniqueFilterType] = None) -> Self:
1038 """Apply unique filtering to the objects returned by this
1039 :class:`_engine.Result`.
1040
1041 When this filter is applied with no arguments, the rows or objects
1042 returned will filtered such that each row is returned uniquely. The
1043 algorithm used to determine this uniqueness is by default the Python
1044 hashing identity of the whole tuple. In some cases a specialized
1045 per-entity hashing scheme may be used, such as when using the ORM, a
1046 scheme is applied which works against the primary key identity of
1047 returned objects.
1048
1049 The unique filter is applied **after all other filters**, which means
1050 if the columns returned have been refined using a method such as the
1051 :meth:`_engine.Result.columns` or :meth:`_engine.Result.scalars`
1052 method, the uniquing is applied to **only the column or columns
1053 returned**. This occurs regardless of the order in which these
1054 methods have been called upon the :class:`_engine.Result` object.
1055
1056 The unique filter also changes the calculus used for methods like
1057 :meth:`_engine.Result.fetchmany` and :meth:`_engine.Result.partitions`.
1058 When using :meth:`_engine.Result.unique`, these methods will continue
1059 to yield the number of rows or objects requested, after uniquing
1060 has been applied. However, this necessarily impacts the buffering
1061 behavior of the underlying cursor or datasource, such that multiple
1062 underlying calls to ``cursor.fetchmany()`` may be necessary in order
1063 to accumulate enough objects in order to provide a unique collection
1064 of the requested size.
1065
1066 :param strategy: a callable that will be applied to rows or objects
1067 being iterated, which should return an object that represents the
1068 unique value of the row. A Python ``set()`` is used to store
1069 these identities. If not passed, a default uniqueness strategy
1070 is used which may have been assembled by the source of this
1071 :class:`_engine.Result` object.
1072
1073 """
1074 self._unique_filter_state = (set(), strategy)
1075 return self
1076
1077 def columns(self, *col_expressions: _KeyIndexType) -> Self:
1078 r"""Establish the columns that should be returned in each row.
1079
1080 This method may be used to limit the columns returned as well
1081 as to reorder them. The given list of expressions are normally
1082 a series of integers or string key names. They may also be
1083 appropriate :class:`.ColumnElement` objects which correspond to
1084 a given statement construct.
1085
1086 .. versionchanged:: 2.0 Due to a bug in 1.4, the
1087 :meth:`_engine.Result.columns` method had an incorrect behavior
1088 where calling upon the method with just one index would cause the
1089 :class:`_engine.Result` object to yield scalar values rather than
1090 :class:`_engine.Row` objects. In version 2.0, this behavior
1091 has been corrected such that calling upon
1092 :meth:`_engine.Result.columns` with a single index will
1093 produce a :class:`_engine.Result` object that continues
1094 to yield :class:`_engine.Row` objects, which include
1095 only a single column.
1096
1097 E.g.::
1098
1099 statement = select(table.c.x, table.c.y, table.c.z)
1100 result = connection.execute(statement)
1101
1102 for z, y in result.columns('z', 'y'):
1103 # ...
1104
1105
1106 Example of using the column objects from the statement itself::
1107
1108 for z, y in result.columns(
1109 statement.selected_columns.c.z,
1110 statement.selected_columns.c.y
1111 ):
1112 # ...
1113
1114 .. versionadded:: 1.4
1115
1116 :param \*col_expressions: indicates columns to be returned. Elements
1117 may be integer row indexes, string column names, or appropriate
1118 :class:`.ColumnElement` objects corresponding to a select construct.
1119
1120 :return: this :class:`_engine.Result` object with the modifications
1121 given.
1122
1123 """
1124 return self._column_slices(col_expressions)
1125
1126 @overload
1127 def scalars(self: Result[Tuple[_T]]) -> ScalarResult[_T]: ...
1128
1129 @overload
1130 def scalars(
1131 self: Result[Tuple[_T]], index: Literal[0]
1132 ) -> ScalarResult[_T]: ...
1133
1134 @overload
1135 def scalars(self, index: _KeyIndexType = 0) -> ScalarResult[Any]: ...
1136
1137 def scalars(self, index: _KeyIndexType = 0) -> ScalarResult[Any]:
1138 """Return a :class:`_engine.ScalarResult` filtering object which
1139 will return single elements rather than :class:`_row.Row` objects.
1140
1141 E.g.::
1142
1143 >>> result = conn.execute(text("select int_id from table"))
1144 >>> result.scalars().all()
1145 [1, 2, 3]
1146
1147 When results are fetched from the :class:`_engine.ScalarResult`
1148 filtering object, the single column-row that would be returned by the
1149 :class:`_engine.Result` is instead returned as the column's value.
1150
1151 .. versionadded:: 1.4
1152
1153 :param index: integer or row key indicating the column to be fetched
1154 from each row, defaults to ``0`` indicating the first column.
1155
1156 :return: a new :class:`_engine.ScalarResult` filtering object referring
1157 to this :class:`_engine.Result` object.
1158
1159 """
1160 return ScalarResult(self, index)
1161
1162 def _getter(
1163 self, key: _KeyIndexType, raiseerr: bool = True
1164 ) -> Optional[Callable[[Row[Any]], Any]]:
1165 """return a callable that will retrieve the given key from a
1166 :class:`_engine.Row`.
1167
1168 """
1169 if self._source_supports_scalars:
1170 raise NotImplementedError(
1171 "can't use this function in 'only scalars' mode"
1172 )
1173 return self._metadata._getter(key, raiseerr)
1174
1175 def _tuple_getter(self, keys: Sequence[_KeyIndexType]) -> _TupleGetterType:
1176 """return a callable that will retrieve the given keys from a
1177 :class:`_engine.Row`.
1178
1179 """
1180 if self._source_supports_scalars:
1181 raise NotImplementedError(
1182 "can't use this function in 'only scalars' mode"
1183 )
1184 return self._metadata._row_as_tuple_getter(keys)
1185
1186 def mappings(self) -> MappingResult:
1187 """Apply a mappings filter to returned rows, returning an instance of
1188 :class:`_engine.MappingResult`.
1189
1190 When this filter is applied, fetching rows will return
1191 :class:`_engine.RowMapping` objects instead of :class:`_engine.Row`
1192 objects.
1193
1194 .. versionadded:: 1.4
1195
1196 :return: a new :class:`_engine.MappingResult` filtering object
1197 referring to this :class:`_engine.Result` object.
1198
1199 """
1200