from ..util import collections_abc
from ..util import py2k
-if util.TYPE_CHECKING:
- from typing import Any
- from typing import List
- from typing import Optional
- from typing import Int
- from typing import Iterator
- from typing import Mapping
-
if _baserow_usecext:
from sqlalchemy.cresultproxy import tuplegetter
return None
if scalar and self._source_supports_scalars:
+ self._generate_rows = False
make_row = None
else:
make_row = self._row_getter
)
else:
next_row = _NO_ROW
-
- if not raise_for_second_row:
# if we checked for second row then that would have
# closed us :)
self._soft_close(hard=True)
@_generative
def unique(self, strategy=None):
- # type: (Optional[object]) -> Result
"""Apply unique filtering to the objects returned by this
:class:`_engine.Result`.
self._unique_filter_state = (set(), strategy)
def columns(self, *col_expressions):
- # type: (*object) -> Result
r"""Establish the columns that should be returned in each row.
This method may be used to limit the columns returned as well
return self._column_slices(col_expressions)
def scalars(self, index=0):
- # type: (Int) -> ScalarResult
"""Return a :class:`_result.ScalarResult` filtering object which
will return single elements rather than :class:`_row.Row` objects.
return self._next_impl()
def partitions(self, size=None):
- # type: (Optional[Int]) -> Iterator[List[Row]]
"""Iterate through sub-lists of rows of the size given.
Each list will be of the size given, excluding the last list to
break
def fetchall(self):
- # type: () -> List[Row]
"""A synonym for the :meth:`_engine.Result.all` method."""
return self._allrows()
def fetchone(self):
- # type: () -> Row
"""Fetch one row.
When all rows are exhausted, returns None.
return row
def fetchmany(self, size=None):
- # type: (Optional[Int]) -> List[Row]
"""Fetch many rows.
When all rows are exhausted, returns an empty list.
return self._manyrow_getter(self, size)
def all(self):
- # type: () -> List[Row]
"""Return all rows in a list.
Closes the result set after invocation. Subsequent invocations
return self._allrows()
def first(self):
- # type: () -> Row
"""Fetch the first row or None if no row is present.
Closes the result set and discards remaining rows.
:meth:`_result.Result.one`
"""
- return self._only_one_row(False, False, False)
+
+ return self._only_one_row(
+ raise_for_second_row=False, raise_for_none=False, scalar=False
+ )
def one_or_none(self):
- # type: () -> Optional[Row]
"""Return at most one result or raise an exception.
Returns ``None`` if the result has no rows.
:meth:`_result.Result.one`
"""
- return self._only_one_row(True, False, False)
+ return self._only_one_row(
+ raise_for_second_row=True, raise_for_none=False, scalar=False
+ )
def scalar_one(self):
- # type: () -> Any
"""Return exactly one scalar result or raise an exception.
This is equivalent to calling :meth:`.Result.scalars` and then
:meth:`.Result.scalars`
"""
- return self._only_one_row(True, True, True)
+ return self._only_one_row(
+ raise_for_second_row=True, raise_for_none=True, scalar=True
+ )
def scalar_one_or_none(self):
- # type: () -> Optional[Any]
"""Return exactly one or no scalar result.
This is equivalent to calling :meth:`.Result.scalars` and then
:meth:`.Result.scalars`
"""
- return self._only_one_row(True, False, True)
+ return self._only_one_row(
+ raise_for_second_row=True, raise_for_none=False, scalar=True
+ )
def one(self):
- # type: () -> Row
"""Return exactly one row or raise an exception.
Raises :class:`.NoResultFound` if the result returns no
:meth:`_result.Result.scalar_one`
"""
- return self._only_one_row(True, True, False)
+ return self._only_one_row(
+ raise_for_second_row=True, raise_for_none=True, scalar=False
+ )
def scalar(self):
- # type: () -> Optional[Any]
"""Fetch the first column of the first row, and close the result set.
Returns None if there are no rows to fetch.
:return: a Python scalar value , or None if no rows remain.
"""
- return self._only_one_row(False, False, True)
+ return self._only_one_row(
+ raise_for_second_row=False, raise_for_none=False, scalar=True
+ )
def freeze(self):
"""Return a callable object that will produce copies of this
self._unique_filter_state = real_result._unique_filter_state
def unique(self, strategy=None):
- # type: () -> ScalarResult
"""Apply unique filtering to the objects returned by this
:class:`_engine.ScalarResult`.
return self
def partitions(self, size=None):
- # type: (Optional[Int]) -> Iterator[List[Any]]
"""Iterate through sub-lists of elements of the size given.
Equivalent to :meth:`_result.Result.partitions` except that
break
def fetchall(self):
- # type: () -> List[Any]
"""A synonym for the :meth:`_engine.ScalarResult.all` method."""
return self._allrows()
def fetchmany(self, size=None):
- # type: (Optional[Int]) -> List[Any]
"""Fetch many objects.
Equivalent to :meth:`_result.Result.fetchmany` except that
return self._manyrow_getter(self, size)
def all(self):
- # type: () -> List[Any]
"""Return all scalar values in a list.
Equivalent to :meth:`_result.Result.all` except that
return self._next_impl()
def first(self):
- # type: () -> Optional[Any]
"""Fetch the first object or None if no object is present.
Equivalent to :meth:`_result.Result.first` except that
"""
- return self._only_one_row(False, False, False)
+ return self._only_one_row(
+ raise_for_second_row=False, raise_for_none=False, scalar=False
+ )
def one_or_none(self):
- # type: () -> Optional[Any]
"""Return at most one object or raise an exception.
Equivalent to :meth:`_result.Result.one_or_none` except that
are returned.
"""
- return self._only_one_row(True, False, False)
+ return self._only_one_row(
+ raise_for_second_row=True, raise_for_none=False, scalar=False
+ )
def one(self):
- # type: () -> Any
"""Return exactly one object or raise an exception.
Equivalent to :meth:`_result.Result.one` except that
are returned.
"""
- return self._only_one_row(True, True, False)
+ return self._only_one_row(
+ raise_for_second_row=True, raise_for_none=True, scalar=False
+ )
class MappingResult(_WithKeys, FilterResult):
self._metadata = self._metadata._reduce([0])
def unique(self, strategy=None):
- # type: () -> MappingResult
"""Apply unique filtering to the objects returned by this
:class:`_engine.MappingResult`.
return self
def columns(self, *col_expressions):
- # type: (*object) -> MappingResult
r"""Establish the columns that should be returned in each row."""
return self._column_slices(col_expressions)
def partitions(self, size=None):
- # type: (Optional[Int]) -> Iterator[List[Mapping]]
"""Iterate through sub-lists of elements of the size given.
Equivalent to :meth:`_result.Result.partitions` except that
break
def fetchall(self):
- # type: () -> List[Mapping]
"""A synonym for the :meth:`_engine.MappingResult.all` method."""
return self._allrows()
def fetchone(self):
- # type: () -> Mapping
"""Fetch one object.
Equivalent to :meth:`_result.Result.fetchone` except that
return row
def fetchmany(self, size=None):
- # type: (Optional[Int]) -> List[Mapping]
"""Fetch many objects.
Equivalent to :meth:`_result.Result.fetchmany` except that
return self._manyrow_getter(self, size)
def all(self):
- # type: () -> List[Mapping]
"""Return all scalar values in a list.
Equivalent to :meth:`_result.Result.all` except that
return self._next_impl()
def first(self):
- # type: () -> Optional[Mapping]
"""Fetch the first object or None if no object is present.
Equivalent to :meth:`_result.Result.first` except that
"""
- return self._only_one_row(False, False, False)
+ return self._only_one_row(
+ raise_for_second_row=False, raise_for_none=False, scalar=False
+ )
def one_or_none(self):
- # type: () -> Optional[Mapping]
"""Return at most one object or raise an exception.
Equivalent to :meth:`_result.Result.one_or_none` except that
are returned.
"""
- return self._only_one_row(True, False, False)
+ return self._only_one_row(
+ raise_for_second_row=True, raise_for_none=False, scalar=False
+ )
def one(self):
- # type: () -> Mapping
"""Return exactly one object or raise an exception.
Equivalent to :meth:`_result.Result.one` except that
are returned.
"""
- return self._only_one_row(True, True, False)
+ return self._only_one_row(
+ raise_for_second_row=True, raise_for_none=True, scalar=False
+ )
class FrozenResult(object):
result = result.scalars(1).unique()
eq_(result.all(), [None, 4])
- def test_scalar_only_on_filter(self):
+ def test_scalar_only_on_filter_w_unique(self):
# test a mixture of the "real" result and the
# scalar filter, where scalar has unique and real result does not.
eq_(next(result), (3, 4, 5)) # non-unique fifth row
eq_(u_s.all(), []) # unique set is done because only 3 is left
- def test_scalar_none_one(self):
+ def test_scalar_none_one_w_unique(self):
result = self._fixture(data=[(1, None, 2)])
result = result.scalars(1).unique()
# one is returning None, see?
eq_(result.one(), None)
- def test_scalar_none_one_or_none(self):
+ def test_scalar_none_one_or_none_w_unique(self):
result = self._fixture(data=[(1, None, 2)])
result = result.scalars(1).unique()
# have to allow for this unforuntately, unless we want to raise?
eq_(result.one_or_none(), None)
+ def test_scalar_one_w_unique(self):
+ result = self._fixture(data=[(1, None, 2)])
+
+ result = result.unique()
+
+ eq_(result.scalar_one(), 1)
+
+ def test_scalars_one_w_unique(self):
+ result = self._fixture(data=[(1, None, 2)])
+
+ result = result.unique()
+
+ eq_(result.scalars().one(), 1)
+
def test_scalar_none_first(self):
result = self._fixture(data=[(1, None, 2)])
eq_(r.all(), [(1,), (2,), (4,)])
+ @testing.combinations(
+ lambda r: r.scalar(),
+ lambda r: r.scalar_one(),
+ lambda r: r.scalar_one_or_none(),
+ argnames="get",
+ )
+ def test_unique_scalar_accessors(self, no_tuple_one_fixture, get):
+ metadata = result.SimpleResultMetaData(
+ ["a", "b", "c"], _unique_filters=[int]
+ )
+
+ r = result.ChunkedIteratorResult(
+ metadata,
+ no_tuple_one_fixture,
+ source_supports_scalars=True,
+ )
+
+ r = r.unique()
+
+ eq_(get(r), 1)
+
def test_scalar_mode_mfiltered_unique_mappings_all(self, no_tuple_fixture):
metadata = result.SimpleResultMetaData(
["a", "b", "c"], _unique_filters=[int]
eq_(list(r), [(1,), (2,), (1,), (1,), (4,)])
- def test_scalar_mode_first(self, no_tuple_one_fixture):
+ @testing.combinations(
+ lambda r: r.one(),
+ lambda r: r.first(),
+ lambda r: r.one_or_none(),
+ argnames="get",
+ )
+ def test_scalar_mode_first(self, no_tuple_one_fixture, get):
metadata = result.SimpleResultMetaData(["a", "b", "c"])
r = result.ChunkedIteratorResult(
metadata, no_tuple_one_fixture, source_supports_scalars=True
)
- eq_(r.one(), (1,))
+ eq_(get(r), (1,))
- def test_scalar_mode_scalar_one(self, no_tuple_one_fixture):
+ @testing.combinations(
+ lambda r: r.scalar(),
+ lambda r: r.scalar_one(),
+ lambda r: r.scalar_one_or_none(),
+ argnames="get",
+ )
+ def test_scalar_mode_scalar_one(self, no_tuple_one_fixture, get):
metadata = result.SimpleResultMetaData(["a", "b", "c"])
r = result.ChunkedIteratorResult(
metadata, no_tuple_one_fixture, source_supports_scalars=True
)
- eq_(r.scalar_one(), 1)
+ eq_(get(r), 1)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
