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