You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
386 lines
11 KiB
386 lines
11 KiB
1 year ago
|
# engine/row.py
|
||
|
# Copyright (C) 2005-2023 the SQLAlchemy authors and contributors
|
||
|
# <see AUTHORS file>
|
||
|
#
|
||
|
# This module is part of SQLAlchemy and is released under
|
||
|
# the MIT License: https://www.opensource.org/licenses/mit-license.php
|
||
|
|
||
|
"""Define row constructs including :class:`.Row`."""
|
||
|
|
||
|
from __future__ import annotations
|
||
|
|
||
|
from abc import ABC
|
||
|
import collections.abc as collections_abc
|
||
|
import operator
|
||
|
import typing
|
||
|
from typing import Any
|
||
|
from typing import Callable
|
||
|
from typing import Dict
|
||
|
from typing import Generic
|
||
|
from typing import Iterator
|
||
|
from typing import List
|
||
|
from typing import Mapping
|
||
|
from typing import NoReturn
|
||
|
from typing import Optional
|
||
|
from typing import overload
|
||
|
from typing import Sequence
|
||
|
from typing import Tuple
|
||
|
from typing import TYPE_CHECKING
|
||
|
from typing import TypeVar
|
||
|
from typing import Union
|
||
|
|
||
|
from ..sql import util as sql_util
|
||
|
from ..util._has_cy import HAS_CYEXTENSION
|
||
|
|
||
|
if TYPE_CHECKING or not HAS_CYEXTENSION:
|
||
|
from ._py_row import BaseRow as BaseRow
|
||
|
from ._py_row import KEY_INTEGER_ONLY
|
||
|
from ._py_row import KEY_OBJECTS_ONLY
|
||
|
else:
|
||
|
from sqlalchemy.cyextension.resultproxy import BaseRow as BaseRow
|
||
|
from sqlalchemy.cyextension.resultproxy import KEY_INTEGER_ONLY
|
||
|
from sqlalchemy.cyextension.resultproxy import KEY_OBJECTS_ONLY
|
||
|
|
||
|
if TYPE_CHECKING:
|
||
|
from .result import _KeyType
|
||
|
from .result import RMKeyView
|
||
|
from ..sql.type_api import _ResultProcessorType
|
||
|
|
||
|
_T = TypeVar("_T", bound=Any)
|
||
|
_TP = TypeVar("_TP", bound=Tuple[Any, ...])
|
||
|
|
||
|
|
||
|
class Row(BaseRow, Sequence[Any], Generic[_TP]):
|
||
|
"""Represent a single result row.
|
||
|
|
||
|
The :class:`.Row` object represents a row of a database result. It is
|
||
|
typically associated in the 1.x series of SQLAlchemy with the
|
||
|
:class:`_engine.CursorResult` object, however is also used by the ORM for
|
||
|
tuple-like results as of SQLAlchemy 1.4.
|
||
|
|
||
|
The :class:`.Row` object seeks to act as much like a Python named
|
||
|
tuple as possible. For mapping (i.e. dictionary) behavior on a row,
|
||
|
such as testing for containment of keys, refer to the :attr:`.Row._mapping`
|
||
|
attribute.
|
||
|
|
||
|
.. seealso::
|
||
|
|
||
|
:ref:`tutorial_selecting_data` - includes examples of selecting
|
||
|
rows from SELECT statements.
|
||
|
|
||
|
.. versionchanged:: 1.4
|
||
|
|
||
|
Renamed ``RowProxy`` to :class:`.Row`. :class:`.Row` is no longer a
|
||
|
"proxy" object in that it contains the final form of data within it,
|
||
|
and now acts mostly like a named tuple. Mapping-like functionality is
|
||
|
moved to the :attr:`.Row._mapping` attribute. See
|
||
|
:ref:`change_4710_core` for background on this change.
|
||
|
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
_default_key_style = KEY_INTEGER_ONLY
|
||
|
|
||
|
def __setattr__(self, name: str, value: Any) -> NoReturn:
|
||
|
raise AttributeError("can't set attribute")
|
||
|
|
||
|
def __delattr__(self, name: str) -> NoReturn:
|
||
|
raise AttributeError("can't delete attribute")
|
||
|
|
||
|
def tuple(self) -> _TP:
|
||
|
"""Return a 'tuple' form of this :class:`.Row`.
|
||
|
|
||
|
At runtime, this method returns "self"; the :class:`.Row` object is
|
||
|
already a named tuple. However, at the typing level, if this
|
||
|
:class:`.Row` is typed, the "tuple" return type will be a :pep:`484`
|
||
|
``Tuple`` datatype that contains typing information about individual
|
||
|
elements, supporting typed unpacking and attribute access.
|
||
|
|
||
|
.. versionadded:: 2.0
|
||
|
|
||
|
.. seealso::
|
||
|
|
||
|
:meth:`.Result.tuples`
|
||
|
|
||
|
"""
|
||
|
return self # type: ignore
|
||
|
|
||
|
@property
|
||
|
def t(self) -> _TP:
|
||
|
"""a synonym for :attr:`.Row.tuple`
|
||
|
|
||
|
.. versionadded:: 2.0
|
||
|
|
||
|
.. seealso::
|
||
|
|
||
|
:meth:`.Result.t`
|
||
|
|
||
|
"""
|
||
|
return self # type: ignore
|
||
|
|
||
|
@property
|
||
|
def _mapping(self) -> RowMapping:
|
||
|
"""Return a :class:`.RowMapping` for this :class:`.Row`.
|
||
|
|
||
|
This object provides a consistent Python mapping (i.e. dictionary)
|
||
|
interface for the data contained within the row. The :class:`.Row`
|
||
|
by itself behaves like a named tuple.
|
||
|
|
||
|
.. seealso::
|
||
|
|
||
|
:attr:`.Row._fields`
|
||
|
|
||
|
.. versionadded:: 1.4
|
||
|
|
||
|
"""
|
||
|
return RowMapping(
|
||
|
self._parent,
|
||
|
None,
|
||
|
self._keymap,
|
||
|
RowMapping._default_key_style,
|
||
|
self._data,
|
||
|
)
|
||
|
|
||
|
def _filter_on_values(
|
||
|
self, filters: Optional[Sequence[Optional[_ResultProcessorType[Any]]]]
|
||
|
) -> Row[Any]:
|
||
|
return Row(
|
||
|
self._parent,
|
||
|
filters,
|
||
|
self._keymap,
|
||
|
self._key_style,
|
||
|
self._data,
|
||
|
)
|
||
|
|
||
|
if not TYPE_CHECKING:
|
||
|
|
||
|
def _special_name_accessor(name: str) -> Any:
|
||
|
"""Handle ambiguous names such as "count" and "index" """
|
||
|
|
||
|
@property
|
||
|
def go(self: Row) -> Any:
|
||
|
if self._parent._has_key(name):
|
||
|
return self.__getattr__(name)
|
||
|
else:
|
||
|
|
||
|
def meth(*arg: Any, **kw: Any) -> Any:
|
||
|
return getattr(collections_abc.Sequence, name)(
|
||
|
self, *arg, **kw
|
||
|
)
|
||
|
|
||
|
return meth
|
||
|
|
||
|
return go
|
||
|
|
||
|
count = _special_name_accessor("count")
|
||
|
index = _special_name_accessor("index")
|
||
|
|
||
|
def __contains__(self, key: Any) -> bool:
|
||
|
return key in self._data
|
||
|
|
||
|
def _op(self, other: Any, op: Callable[[Any, Any], bool]) -> bool:
|
||
|
return (
|
||
|
op(tuple(self), tuple(other))
|
||
|
if isinstance(other, Row)
|
||
|
else op(tuple(self), other)
|
||
|
)
|
||
|
|
||
|
__hash__ = BaseRow.__hash__
|
||
|
|
||
|
if TYPE_CHECKING:
|
||
|
|
||
|
@overload
|
||
|
def __getitem__(self, index: int) -> Any:
|
||
|
...
|
||
|
|
||
|
@overload
|
||
|
def __getitem__(self, index: slice) -> Sequence[Any]:
|
||
|
...
|
||
|
|
||
|
def __getitem__(
|
||
|
self, index: Union[int, slice]
|
||
|
) -> Union[Any, Sequence[Any]]:
|
||
|
...
|
||
|
|
||
|
def __lt__(self, other: Any) -> bool:
|
||
|
return self._op(other, operator.lt)
|
||
|
|
||
|
def __le__(self, other: Any) -> bool:
|
||
|
return self._op(other, operator.le)
|
||
|
|
||
|
def __ge__(self, other: Any) -> bool:
|
||
|
return self._op(other, operator.ge)
|
||
|
|
||
|
def __gt__(self, other: Any) -> bool:
|
||
|
return self._op(other, operator.gt)
|
||
|
|
||
|
def __eq__(self, other: Any) -> bool:
|
||
|
return self._op(other, operator.eq)
|
||
|
|
||
|
def __ne__(self, other: Any) -> bool:
|
||
|
return self._op(other, operator.ne)
|
||
|
|
||
|
def __repr__(self) -> str:
|
||
|
return repr(sql_util._repr_row(self))
|
||
|
|
||
|
@property
|
||
|
def _fields(self) -> Tuple[str, ...]:
|
||
|
"""Return a tuple of string keys as represented by this
|
||
|
:class:`.Row`.
|
||
|
|
||
|
The keys can represent the labels of the columns returned by a core
|
||
|
statement or the names of the orm classes returned by an orm
|
||
|
execution.
|
||
|
|
||
|
This attribute is analogous to the Python named tuple ``._fields``
|
||
|
attribute.
|
||
|
|
||
|
.. versionadded:: 1.4
|
||
|
|
||
|
.. seealso::
|
||
|
|
||
|
:attr:`.Row._mapping`
|
||
|
|
||
|
"""
|
||
|
return tuple([k for k in self._parent.keys if k is not None])
|
||
|
|
||
|
def _asdict(self) -> Dict[str, Any]:
|
||
|
"""Return a new dict which maps field names to their corresponding
|
||
|
values.
|
||
|
|
||
|
This method is analogous to the Python named tuple ``._asdict()``
|
||
|
method, and works by applying the ``dict()`` constructor to the
|
||
|
:attr:`.Row._mapping` attribute.
|
||
|
|
||
|
.. versionadded:: 1.4
|
||
|
|
||
|
.. seealso::
|
||
|
|
||
|
:attr:`.Row._mapping`
|
||
|
|
||
|
"""
|
||
|
return dict(self._mapping)
|
||
|
|
||
|
|
||
|
BaseRowProxy = BaseRow
|
||
|
RowProxy = Row
|
||
|
|
||
|
|
||
|
class ROMappingView(ABC):
|
||
|
__slots__ = ()
|
||
|
|
||
|
_items: Sequence[Any]
|
||
|
_mapping: Mapping[str, Any]
|
||
|
|
||
|
def __init__(self, mapping: Mapping[str, Any], items: Sequence[Any]):
|
||
|
self._mapping = mapping
|
||
|
self._items = items
|
||
|
|
||
|
def __len__(self) -> int:
|
||
|
return len(self._items)
|
||
|
|
||
|
def __repr__(self) -> str:
|
||
|
return "{0.__class__.__name__}({0._mapping!r})".format(self)
|
||
|
|
||
|
def __iter__(self) -> Iterator[Any]:
|
||
|
return iter(self._items)
|
||
|
|
||
|
def __contains__(self, item: Any) -> bool:
|
||
|
return item in self._items
|
||
|
|
||
|
def __eq__(self, other: Any) -> bool:
|
||
|
return list(other) == list(self)
|
||
|
|
||
|
def __ne__(self, other: Any) -> bool:
|
||
|
return list(other) != list(self)
|
||
|
|
||
|
|
||
|
class ROMappingKeysValuesView(
|
||
|
ROMappingView, typing.KeysView[str], typing.ValuesView[Any]
|
||
|
):
|
||
|
__slots__ = ("_items",)
|
||
|
|
||
|
|
||
|
class ROMappingItemsView(ROMappingView, typing.ItemsView[str, Any]):
|
||
|
__slots__ = ("_items",)
|
||
|
|
||
|
|
||
|
class RowMapping(BaseRow, typing.Mapping[str, Any]):
|
||
|
"""A ``Mapping`` that maps column names and objects to :class:`.Row`
|
||
|
values.
|
||
|
|
||
|
The :class:`.RowMapping` is available from a :class:`.Row` via the
|
||
|
:attr:`.Row._mapping` attribute, as well as from the iterable interface
|
||
|
provided by the :class:`.MappingResult` object returned by the
|
||
|
:meth:`_engine.Result.mappings` method.
|
||
|
|
||
|
:class:`.RowMapping` supplies Python mapping (i.e. dictionary) access to
|
||
|
the contents of the row. This includes support for testing of
|
||
|
containment of specific keys (string column names or objects), as well
|
||
|
as iteration of keys, values, and items::
|
||
|
|
||
|
for row in result:
|
||
|
if 'a' in row._mapping:
|
||
|
print("Column 'a': %s" % row._mapping['a'])
|
||
|
|
||
|
print("Column b: %s" % row._mapping[table.c.b])
|
||
|
|
||
|
|
||
|
.. versionadded:: 1.4 The :class:`.RowMapping` object replaces the
|
||
|
mapping-like access previously provided by a database result row,
|
||
|
which now seeks to behave mostly like a named tuple.
|
||
|
|
||
|
"""
|
||
|
|
||
|
__slots__ = ()
|
||
|
|
||
|
_default_key_style = KEY_OBJECTS_ONLY
|
||
|
|
||
|
if TYPE_CHECKING:
|
||
|
|
||
|
def __getitem__(self, key: _KeyType) -> Any:
|
||
|
...
|
||
|
|
||
|
else:
|
||
|
__getitem__ = BaseRow._get_by_key_impl_mapping
|
||
|
|
||
|
def _values_impl(self) -> List[Any]:
|
||
|
return list(self._data)
|
||
|
|
||
|
def __iter__(self) -> Iterator[str]:
|
||
|
return (k for k in self._parent.keys if k is not None)
|
||
|
|
||
|
def __len__(self) -> int:
|
||
|
return len(self._data)
|
||
|
|
||
|
def __contains__(self, key: object) -> bool:
|
||
|
return self._parent._has_key(key)
|
||
|
|
||
|
def __repr__(self) -> str:
|
||
|
return repr(dict(self))
|
||
|
|
||
|
def items(self) -> ROMappingItemsView:
|
||
|
"""Return a view of key/value tuples for the elements in the
|
||
|
underlying :class:`.Row`.
|
||
|
|
||
|
"""
|
||
|
return ROMappingItemsView(
|
||
|
self, [(key, self[key]) for key in self.keys()]
|
||
|
)
|
||
|
|
||
|
def keys(self) -> RMKeyView:
|
||
|
"""Return a view of 'keys' for string column names represented
|
||
|
by the underlying :class:`.Row`.
|
||
|
|
||
|
"""
|
||
|
|
||
|
return self._parent.keys
|
||
|
|
||
|
def values(self) -> ROMappingKeysValuesView:
|
||
|
"""Return a view of values for the values represented in the
|
||
|
underlying :class:`.Row`.
|
||
|
|
||
|
"""
|
||
|
return ROMappingKeysValuesView(self, self._values_impl())
|