@overload
def scalar(
self,
- statement: TypedReturnsRows[Tuple[_T]],
+ statement: TypedReturnsRows[_T],
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
execution_options: Optional[CoreExecuteOptionsParameter] = None,
@overload
def scalars(
self,
- statement: TypedReturnsRows[Tuple[_T]],
+ statement: TypedReturnsRows[_T],
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
execution_options: Optional[CoreExecuteOptionsParameter] = None,
@overload
def execute(
self,
- statement: TypedReturnsRows[Tuple[Unpack[_Ts]]],
+ statement: TypedReturnsRows[Unpack[_Ts]],
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
execution_options: Optional[CoreExecuteOptionsParameter] = None,
@overload
def scalars(
- self: AsyncResult[Tuple[_T]], index: Literal[0]
+ self: AsyncResult[_T, Unpack[Tuple[Any, ...]]], index: Literal[0]
) -> AsyncScalarResult[_T]:
...
@overload
- def scalars(self: AsyncResult[Tuple[_T]]) -> AsyncScalarResult[_T]:
+ def scalars(
+ self: AsyncResult[_T, Unpack[Tuple[Any, ...]]],
+ ) -> AsyncScalarResult[_T]:
...
@overload
from ...util import ScopedRegistry
from ...util import warn
from ...util import warn_deprecated
+from ...util.typing import TypeVarTuple
from ...util.typing import Unpack
if TYPE_CHECKING:
from ...sql.selectable import TypedReturnsRows
_T = TypeVar("_T", bound=Any)
+_Ts = TypeVarTuple("_Ts")
@create_proxy_methods(
@overload
async def execute(
self,
- statement: TypedReturnsRows[_T],
+ statement: TypedReturnsRows[Tuple[Unpack[_Ts]]],
params: Optional[_CoreAnyExecuteParams] = None,
*,
execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
- ) -> Result[_T]:
+ ) -> Result[Unpack[_Ts]]:
...
@overload
from ...orm import SessionTransaction
from ...orm import state as _instance_state
from ...util.concurrency import greenlet_spawn
+from ...util.typing import TypeVarTuple
from ...util.typing import Unpack
+
if TYPE_CHECKING:
from .engine import AsyncConnection
from .engine import AsyncEngine
_AsyncSessionBind = Union["AsyncEngine", "AsyncConnection"]
_T = TypeVar("_T", bound=Any)
-
+_Ts = TypeVarTuple("_Ts")
_EXECUTE_OPTIONS = util.immutabledict({"prebuffer_rows": True})
_STREAM_OPTIONS = util.immutabledict({"stream_results": True})
@overload
async def execute(
self,
- statement: TypedReturnsRows[_T],
+ statement: TypedReturnsRows[Tuple[Unpack[_Ts]]],
params: Optional[_CoreAnyExecuteParams] = None,
*,
execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
- ) -> Result[_T]:
+ ) -> Result[Unpack[_Ts]]:
...
@overload
from ..sql import roles
from ..sql import util as sql_util
from ..sql import visitors
-from ..sql._typing import _TP
from ..sql._typing import is_dml
from ..sql._typing import is_insert_update
from ..sql._typing import is_select_base
from ..sql.selectable import SelectState
from ..sql.selectable import TypedReturnsRows
from ..sql.visitors import InternalTraversal
+from ..util.typing import TypeVarTuple
+from ..util.typing import Unpack
+
if TYPE_CHECKING:
from ._typing import _InternalEntityType
from ..sql.type_api import TypeEngine
_T = TypeVar("_T", bound=Any)
+_Ts = TypeVarTuple("_Ts")
_path_registry = PathRegistry.root
_EMPTY_DICT = util.immutabledict()
def __init__(
self,
compile_state: CompileState,
- statement: Union[Select[Any], FromStatement[Any]],
+ statement: Union[
+ Select[Unpack[Tuple[Any, ...]]],
+ FromStatement[Unpack[Tuple[Any, ...]]],
+ ],
params: _CoreSingleExecuteParams,
session: Session,
load_options: Union[
attributes: Dict[Any, Any]
global_attributes: Dict[Any, Any]
- statement: Union[Select[Any], FromStatement[Any]]
- select_statement: Union[Select[Any], FromStatement[Any]]
+ statement: Union[
+ Select[Unpack[Tuple[Any, ...]]], FromStatement[Unpack[Tuple[Any, ...]]]
+ ]
+ select_statement: Union[
+ Select[Unpack[Tuple[Any, ...]]], FromStatement[Unpack[Tuple[Any, ...]]]
+ ]
_entities: List[_QueryEntity]
_polymorphic_adapters: Dict[_InternalEntityType, ORMAdapter]
compile_options: Union[
entity.setup_dml_returning_compile_state(self, adapter)
-class FromStatement(GroupedElement, Generative, TypedReturnsRows[_TP]):
+class FromStatement(GroupedElement, Generative, TypedReturnsRows[Unpack[_Ts]]):
"""Core construct that represents a load of ORM objects from various
:class:`.ReturnsRows` and other classes including:
def _legacy_filter_by_entity_zero(
- query_or_augmented_select: Union[Query[Any], Select[Any]]
+ query_or_augmented_select: Union[
+ Query[Any], Select[Unpack[Tuple[Any, ...]]]
+ ]
) -> Optional[_InternalEntityType[Any]]:
self = query_or_augmented_select
if self._setup_joins:
def _entity_from_pre_ent_zero(
- query_or_augmented_select: Union[Query[Any], Select[Any]]
+ query_or_augmented_select: Union[
+ Query[Any], Select[Unpack[Tuple[Any, ...]]]
+ ]
) -> Optional[_InternalEntityType[Any]]:
self = query_or_augmented_select
if not self._raw_columns:
def create_row_processor(
self,
- query: Select[Any],
+ query: Select[Unpack[Tuple[Any, ...]]],
procs: Sequence[Callable[[Row[Unpack[Tuple[Any, ...]]]], Any]],
labels: Sequence[str],
) -> Callable[[Row[Unpack[Tuple[Any, ...]]]], Any]:
from ..sql import util as sql_util
from ..sql import visitors
from ..sql._typing import _FromClauseArgument
-from ..sql._typing import _TP
from ..sql.annotation import SupportsCloneAnnotations
from ..sql.base import _entity_namespace_key
from ..sql.base import _generative
from ..sql.selectable import SelectLabelStyle
from ..util.typing import Literal
from ..util.typing import Self
+from ..util.typing import TypeVarTuple
+from ..util.typing import Unpack
if TYPE_CHECKING:
__all__ = ["Query", "QueryContext"]
_T = TypeVar("_T", bound=Any)
+_Ts = TypeVarTuple("_Ts")
@inspection._self_inspects
return stmt
- def _final_statement(self, legacy_query_style: bool = True) -> Select[Any]:
+ def _final_statement(
+ self, legacy_query_style: bool = True
+ ) -> Select[Unpack[Tuple[Any, ...]]]:
"""Return the 'final' SELECT statement for this :class:`.Query`.
This is used by the testing suite only and is fairly inefficient.
@overload
def only_return_tuples(
self: Query[_O], value: Literal[True]
- ) -> RowReturningQuery[Tuple[_O]]:
+ ) -> RowReturningQuery[_O]:
...
@overload
@overload
def with_entities(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1]
- ) -> RowReturningQuery[Tuple[_T0, _T1]]:
+ ) -> RowReturningQuery[_T0, _T1]:
...
@overload
def with_entities(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1], __ent2: _TCCA[_T2]
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2]:
...
@overload
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3]:
...
@overload
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4]:
...
@overload
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4, _T5]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4, _T5]:
...
@overload
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4, _T5, _T6]:
...
@overload
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
__ent7: _TCCA[_T7],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]:
...
# END OVERLOADED FUNCTIONS self.with_entities
"""BulkUD which handles DELETEs."""
-class RowReturningQuery(Query[Row[_TP]]):
+class RowReturningQuery(Query[Row[Unpack[_Ts]]]):
if TYPE_CHECKING:
- def tuples(self) -> Query[_TP]: # type: ignore
+ def tuples(self) -> Query[tuple[Unpack[_Ts]]]: # type: ignore
...
@overload
def query(
self, _colexpr: TypedColumnsClauseRole[_T]
- ) -> RowReturningQuery[Tuple[_T]]:
+ ) -> RowReturningQuery[_T]:
...
# START OVERLOADED FUNCTIONS self.query RowReturningQuery 2-8
@overload
def query(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1]
- ) -> RowReturningQuery[Tuple[_T0, _T1]]:
+ ) -> RowReturningQuery[_T0, _T1]:
...
@overload
def query(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1], __ent2: _TCCA[_T2]
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2]:
...
@overload
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3]:
...
@overload
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4]:
...
@overload
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4, _T5]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4, _T5]:
...
@overload
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4, _T5, _T6]:
...
@overload
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
__ent7: _TCCA[_T7],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]:
...
# END OVERLOADED FUNCTIONS self.query
@overload
def query(
self, _colexpr: TypedColumnsClauseRole[_T]
- ) -> RowReturningQuery[Tuple[_T]]:
+ ) -> RowReturningQuery[_T]:
...
# START OVERLOADED FUNCTIONS self.query RowReturningQuery 2-8
@overload
def query(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1]
- ) -> RowReturningQuery[Tuple[_T0, _T1]]:
+ ) -> RowReturningQuery[_T0, _T1]:
...
@overload
def query(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1], __ent2: _TCCA[_T2]
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2]:
...
@overload
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3]:
...
@overload
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4]:
...
@overload
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4, _T5]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4, _T5]:
...
@overload
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4, _T5, _T6]:
...
@overload
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
__ent7: _TCCA[_T7],
- ) -> RowReturningQuery[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]]:
+ ) -> RowReturningQuery[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]:
...
# END OVERLOADED FUNCTIONS self.query
with_for_update = ForUpdateArg._from_argument(with_for_update)
- stmt: Select[Any] = sql.select(object_mapper(instance))
+ stmt: Select[Unpack[Tuple[Any, ...]]] = sql.select(
+ object_mapper(instance)
+ )
if (
loading.load_on_ident(
self,
def create_row_processor(
self,
- query: Select[Any],
+ query: Select[Unpack[Tuple[Any, ...]]],
procs: Sequence[Callable[[Row[Unpack[Tuple[Any, ...]]]], Any]],
labels: Sequence[str],
) -> Callable[[Row[Unpack[Tuple[Any, ...]]]], Any]:
from .selectable import TableClause
from .selectable import TableSample
from .selectable import Values
+from ..util.typing import Unpack
if TYPE_CHECKING:
from ._typing import _FromClauseArgument
@overload
-def select(__ent0: _TCCA[_T0]) -> Select[Tuple[_T0]]:
+def select(__ent0: _TCCA[_T0]) -> Select[_T0]:
...
@overload
-def select(__ent0: _TCCA[_T0], __ent1: _TCCA[_T1]) -> Select[Tuple[_T0, _T1]]:
+def select(__ent0: _TCCA[_T0], __ent1: _TCCA[_T1]) -> Select[_T0, _T1]:
...
@overload
def select(
__ent0: _TCCA[_T0], __ent1: _TCCA[_T1], __ent2: _TCCA[_T2]
-) -> Select[Tuple[_T0, _T1, _T2]]:
+) -> Select[_T0, _T1, _T2]:
...
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
-) -> Select[Tuple[_T0, _T1, _T2, _T3]]:
+) -> Select[_T0, _T1, _T2, _T3]:
...
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
-) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4]]:
+) -> Select[_T0, _T1, _T2, _T3, _T4]:
...
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
-) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5]]:
+) -> Select[_T0, _T1, _T2, _T3, _T4, _T5]:
...
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
-) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6]]:
+) -> Select[_T0, _T1, _T2, _T3, _T4, _T5, _T6]:
...
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
__ent7: _TCCA[_T7],
-) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]]:
+) -> Select[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]:
...
__ent6: _TCCA[_T6],
__ent7: _TCCA[_T7],
__ent8: _TCCA[_T8],
-) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7, _T8]]:
+) -> Select[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7, _T8]:
...
__ent7: _TCCA[_T7],
__ent8: _TCCA[_T8],
__ent9: _TCCA[_T9],
-) -> Select[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7, _T8, _T9]]:
+) -> Select[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7, _T8, _T9]:
...
@overload
-def select(*entities: _ColumnsClauseArgument[Any], **__kw: Any) -> Select[Any]:
+def select(
+ *entities: _ColumnsClauseArgument[Any], **__kw: Any
+) -> Select[Unpack[Tuple[Any, ...]]]:
...
-def select(*entities: _ColumnsClauseArgument[Any], **__kw: Any) -> Select[Any]:
+def select(
+ *entities: _ColumnsClauseArgument[Any], **__kw: Any
+) -> Select[Unpack[Tuple[Any, ...]]]:
r"""Construct a new :class:`_expression.Select`.
from ..util.typing import Literal
from ..util.typing import Protocol
from ..util.typing import TypeAlias
+from ..util.typing import Unpack
if TYPE_CHECKING:
from datetime import date
def is_select_statement(
t: Union[Executable, ReturnsRows]
- ) -> TypeGuard[Select[Any]]:
+ ) -> TypeGuard[Select[Unpack[Tuple[Any, ...]]]]:
...
def is_table(t: FromClause) -> TypeGuard[TableClause]:
from ..util.typing import Literal
from ..util.typing import Protocol
from ..util.typing import TypedDict
+from ..util.typing import Unpack
if typing.TYPE_CHECKING:
from .annotation import _AnnotationDict
need_result_map_for_nested: bool
need_result_map_for_compound: bool
select_0: ReturnsRows
- insert_from_select: Select[Any]
+ insert_from_select: Select[Unpack[Tuple[Any, ...]]]
class ExpandedState(NamedTuple):
return text
def _setup_select_hints(
- self, select: Select[Any]
+ self, select: Select[Unpack[Tuple[Any, ...]]]
) -> Tuple[str, _FromHintsType]:
byfrom = {
from_: hinttext
from . import coercions
from . import roles
from . import util as sql_util
-from ._typing import _TP
from ._typing import _unexpected_kw
from ._typing import is_column_element
from ._typing import is_named_from_clause
from .. import util
from ..util.typing import Self
from ..util.typing import TypeGuard
+from ..util.typing import TypeVarTuple
+from ..util.typing import Unpack
+
if TYPE_CHECKING:
from ._typing import _ColumnExpressionArgument
_T = TypeVar("_T", bound=Any)
+_Ts = TypeVarTuple("_Ts")
_DMLColumnElement = Union[str, ColumnClause[Any]]
_DMLTableElement = Union[TableClause, Alias, Join]
_supports_multi_parameters = False
- select: Optional[Select[Any]] = None
+ select: Optional[Select[Unpack[Tuple[Any, ...]]]] = None
"""SELECT statement for INSERT .. FROM SELECT"""
_post_values_clause: Optional[ClauseElement] = None
@overload
def returning(
self, __ent0: _TCCA[_T0], *, sort_by_parameter_order: bool = False
- ) -> ReturningInsert[Tuple[_T0]]:
+ ) -> ReturningInsert[_T0]:
...
@overload
__ent1: _TCCA[_T1],
*,
sort_by_parameter_order: bool = False,
- ) -> ReturningInsert[Tuple[_T0, _T1]]:
+ ) -> ReturningInsert[_T0, _T1]:
...
@overload
__ent2: _TCCA[_T2],
*,
sort_by_parameter_order: bool = False,
- ) -> ReturningInsert[Tuple[_T0, _T1, _T2]]:
+ ) -> ReturningInsert[_T0, _T1, _T2]:
...
@overload
__ent3: _TCCA[_T3],
*,
sort_by_parameter_order: bool = False,
- ) -> ReturningInsert[Tuple[_T0, _T1, _T2, _T3]]:
+ ) -> ReturningInsert[_T0, _T1, _T2, _T3]:
...
@overload
__ent4: _TCCA[_T4],
*,
sort_by_parameter_order: bool = False,
- ) -> ReturningInsert[Tuple[_T0, _T1, _T2, _T3, _T4]]:
+ ) -> ReturningInsert[_T0, _T1, _T2, _T3, _T4]:
...
@overload
__ent5: _TCCA[_T5],
*,
sort_by_parameter_order: bool = False,
- ) -> ReturningInsert[Tuple[_T0, _T1, _T2, _T3, _T4, _T5]]:
+ ) -> ReturningInsert[_T0, _T1, _T2, _T3, _T4, _T5]:
...
@overload
__ent6: _TCCA[_T6],
*,
sort_by_parameter_order: bool = False,
- ) -> ReturningInsert[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6]]:
+ ) -> ReturningInsert[_T0, _T1, _T2, _T3, _T4, _T5, _T6]:
...
@overload
__ent7: _TCCA[_T7],
*,
sort_by_parameter_order: bool = False,
- ) -> ReturningInsert[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]]:
+ ) -> ReturningInsert[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]:
...
# END OVERLOADED FUNCTIONS self.returning
...
-class ReturningInsert(Insert, TypedReturnsRows[_TP]):
+class ReturningInsert(Insert, TypedReturnsRows[Unpack[_Ts]]):
"""Typing-only class that establishes a generic type form of
:class:`.Insert` which tracks returned column types.
# statically generated** by tools/generate_tuple_map_overloads.py
@overload
- def returning(self, __ent0: _TCCA[_T0]) -> ReturningUpdate[Tuple[_T0]]:
+ def returning(self, __ent0: _TCCA[_T0]) -> ReturningUpdate[_T0]:
...
@overload
def returning(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1]
- ) -> ReturningUpdate[Tuple[_T0, _T1]]:
+ ) -> ReturningUpdate[_T0, _T1]:
...
@overload
def returning(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1], __ent2: _TCCA[_T2]
- ) -> ReturningUpdate[Tuple[_T0, _T1, _T2]]:
+ ) -> ReturningUpdate[_T0, _T1, _T2]:
...
@overload
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
- ) -> ReturningUpdate[Tuple[_T0, _T1, _T2, _T3]]:
+ ) -> ReturningUpdate[_T0, _T1, _T2, _T3]:
...
@overload
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
- ) -> ReturningUpdate[Tuple[_T0, _T1, _T2, _T3, _T4]]:
+ ) -> ReturningUpdate[_T0, _T1, _T2, _T3, _T4]:
...
@overload
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
- ) -> ReturningUpdate[Tuple[_T0, _T1, _T2, _T3, _T4, _T5]]:
+ ) -> ReturningUpdate[_T0, _T1, _T2, _T3, _T4, _T5]:
...
@overload
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
- ) -> ReturningUpdate[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6]]:
+ ) -> ReturningUpdate[_T0, _T1, _T2, _T3, _T4, _T5, _T6]:
...
@overload
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
__ent7: _TCCA[_T7],
- ) -> ReturningUpdate[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]]:
+ ) -> ReturningUpdate[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]:
...
# END OVERLOADED FUNCTIONS self.returning
...
-class ReturningUpdate(Update, TypedReturnsRows[_TP]):
+class ReturningUpdate(Update, TypedReturnsRows[Unpack[_Ts]]):
"""Typing-only class that establishes a generic type form of
:class:`.Update` which tracks returned column types.
# statically generated** by tools/generate_tuple_map_overloads.py
@overload
- def returning(self, __ent0: _TCCA[_T0]) -> ReturningDelete[Tuple[_T0]]:
+ def returning(self, __ent0: _TCCA[_T0]) -> ReturningDelete[_T0]:
...
@overload
def returning(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1]
- ) -> ReturningDelete[Tuple[_T0, _T1]]:
+ ) -> ReturningDelete[_T0, _T1]:
...
@overload
def returning(
self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1], __ent2: _TCCA[_T2]
- ) -> ReturningDelete[Tuple[_T0, _T1, _T2]]:
+ ) -> ReturningDelete[_T0, _T1, _T2]:
...
@overload
__ent1: _TCCA[_T1],
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
- ) -> ReturningDelete[Tuple[_T0, _T1, _T2, _T3]]:
+ ) -> ReturningDelete[_T0, _T1, _T2, _T3]:
...
@overload
__ent2: _TCCA[_T2],
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
- ) -> ReturningDelete[Tuple[_T0, _T1, _T2, _T3, _T4]]:
+ ) -> ReturningDelete[_T0, _T1, _T2, _T3, _T4]:
...
@overload
__ent3: _TCCA[_T3],
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
- ) -> ReturningDelete[Tuple[_T0, _T1, _T2, _T3, _T4, _T5]]:
+ ) -> ReturningDelete[_T0, _T1, _T2, _T3, _T4, _T5]:
...
@overload
__ent4: _TCCA[_T4],
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
- ) -> ReturningDelete[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6]]:
+ ) -> ReturningDelete[_T0, _T1, _T2, _T3, _T4, _T5, _T6]:
...
@overload
__ent5: _TCCA[_T5],
__ent6: _TCCA[_T6],
__ent7: _TCCA[_T7],
- ) -> ReturningDelete[Tuple[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]]:
+ ) -> ReturningDelete[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]:
...
# END OVERLOADED FUNCTIONS self.returning
@overload
def returning(
self, *cols: _ColumnsClauseArgument[Any], **__kw: Any
- ) -> ReturningDelete[Any]:
+ ) -> ReturningDelete[Unpack[Tuple[Any, ...]]]:
...
def returning(
self, *cols: _ColumnsClauseArgument[Any], **__kw: Any
- ) -> ReturningDelete[Any]:
+ ) -> ReturningDelete[Unpack[Tuple[Any, ...]]]:
...
-class ReturningDelete(Update, TypedReturnsRows[_TP]):
+class ReturningDelete(Update, TypedReturnsRows[Unpack[_Ts]]):
"""Typing-only class that establishes a generic type form of
:class:`.Delete` which tracks returned column types.
from .type_api import TypeEngine
from .visitors import InternalTraversal
from .. import util
+from ..util.typing import Unpack
if TYPE_CHECKING:
joins_implicitly=joins_implicitly,
)
- def select(self) -> Select[Any]:
+ def select(self) -> Select[Unpack[Tuple[Any, ...]]]:
"""Produce a :func:`_expression.select` construct
against this :class:`.FunctionElement`.
s = select(function_element)
"""
- s: Select[Any] = Select(self)
+ s: Select[Unpack[Tuple[Any, ...]]] = Select(self)
if self._execution_options:
s = s.execution_options(**self._execution_options)
return s
from . import visitors
from ._typing import _ColumnsClauseArgument
from ._typing import _no_kw
-from ._typing import _TP
from ._typing import is_column_element
from ._typing import is_select_statement
from ._typing import is_subquery
from ..util.typing import Literal
from ..util.typing import Protocol
from ..util.typing import Self
+from ..util.typing import TypeVarTuple
+from ..util.typing import Unpack
+
and_ = BooleanClauseList.and_
_T = TypeVar("_T", bound=Any)
+_Ts = TypeVarTuple("_Ts")
+
if TYPE_CHECKING:
from ._typing import _ColumnExpressionArgument
"""base for executable statements that return rows."""
-class TypedReturnsRows(ExecutableReturnsRows, Generic[_TP]):
+class TypedReturnsRows(ExecutableReturnsRows, Generic[Unpack[_Ts]]):
"""base for executable statements that return rows."""
_use_schema_map = False
- def select(self) -> Select[Any]:
+ def select(self) -> Select[Unpack[Tuple[Any, ...]]]:
r"""Return a SELECT of this :class:`_expression.FromClause`.
"join explicitly." % (a.description, b.description)
)
- def select(self) -> Select[Any]:
+ def select(self) -> Select[Unpack[Tuple[Any, ...]]]:
r"""Create a :class:`_expression.Select` from this
:class:`_expression.Join`.
def _init(
self,
- selectable: Select[Any],
+ selectable: Select[Unpack[Tuple[Any, ...]]],
*,
name: Optional[str] = None,
recursive: bool = False,
"first in order to create "
"a subquery, which then can be selected.",
)
- def select(self, *arg: Any, **kw: Any) -> Select[Any]:
+ def select(self, *arg: Any, **kw: Any) -> Select[Unpack[Tuple[Any, ...]]]:
return self._implicit_subquery.select(*arg, **kw)
@HasMemoized.memoized_attribute
def __init__(
self,
- statement: Select[Any],
+ statement: Select[Unpack[Tuple[Any, ...]]],
compiler: Optional[SQLCompiler],
**kw: Any,
):
@classmethod
def get_column_descriptions(
- cls, statement: Select[Any]
+ cls, statement: Select[Unpack[Tuple[Any, ...]]]
) -> List[Dict[str, Any]]:
return [
{
@classmethod
def from_statement(
- cls, statement: Select[Any], from_statement: roles.ReturnsRowsRole
+ cls,
+ statement: Select[Unpack[Tuple[Any, ...]]],
+ from_statement: roles.ReturnsRowsRole,
) -> ExecutableReturnsRows:
cls._plugin_not_implemented()
@classmethod
def get_columns_clause_froms(
- cls, statement: Select[Any]
+ cls, statement: Select[Unpack[Tuple[Any, ...]]]
) -> List[FromClause]:
return cls._normalize_froms(
itertools.chain.from_iterable(
return go
- def _get_froms(self, statement: Select[Any]) -> List[FromClause]:
+ def _get_froms(
+ self, statement: Select[Unpack[Tuple[Any, ...]]]
+ ) -> List[FromClause]:
ambiguous_table_name_map: _AmbiguousTableNameMap
self._ambiguous_table_name_map = ambiguous_table_name_map = {}
def _normalize_froms(
cls,
iterable_of_froms: Iterable[FromClause],
- check_statement: Optional[Select[Any]] = None,
+ check_statement: Optional[Select[Unpack[Tuple[Any, ...]]]] = None,
ambiguous_table_name_map: Optional[_AmbiguousTableNameMap] = None,
) -> List[FromClause]:
"""given an iterable of things to select FROM, reduce them to what
@classmethod
def determine_last_joined_entity(
- cls, stmt: Select[Any]
+ cls, stmt: Select[Unpack[Tuple[Any, ...]]]
) -> Optional[_JoinTargetElement]:
if stmt._setup_joins:
return stmt._setup_joins[-1][0]
return None
@classmethod
- def all_selected_columns(cls, statement: Select[Any]) -> _SelectIterable:
+ def all_selected_columns(
+ cls, statement: Select[Unpack[Tuple[Any, ...]]]
+ ) -> _SelectIterable:
return [c for c in _select_iterables(statement._raw_columns)]
def _setup_joins(
return c
@classmethod
- def _generate_for_statement(cls, select_stmt: Select[Any]) -> None:
+ def _generate_for_statement(
+ cls, select_stmt: Select[Unpack[Tuple[Any, ...]]]
+ ) -> None:
if select_stmt._setup_joins or select_stmt._with_options:
self = _MemoizedSelectEntities()
self._raw_columns = select_stmt._raw_columns
HasCompileState,
_SelectFromElements,
GenerativeSelect,
- TypedReturnsRows[_TP],
+ TypedReturnsRows[Unpack[_Ts]],
):
"""Represents a ``SELECT`` statement.
_compile_state_factory: Type[SelectState]
@classmethod
- def _create_raw_select(cls, **kw: Any) -> Select[Any]:
+ def _create_raw_select(cls, **kw: Any) -> Select[Unpack[Tuple[Any, ...]]]:
"""Create a :class:`.Select` using raw ``__new__`` with no coercions.
Used internally to build up :class:`.Select` constructs with
@overload
def scalar_subquery(
- self: Select[Tuple[_MAYBE_ENTITY]],
+ self: Select[_MAYBE_ENTITY],
) -> ScalarSelect[Any]:
...
@overload
def scalar_subquery(
- self: Select[Tuple[_NOT_ENTITY]],
+ self: Select[_NOT_ENTITY],
) -> ScalarSelect[_NOT_ENTITY]:
...
@_generative
def add_columns(
self, *entities: _ColumnsClauseArgument[Any]
- ) -> Select[Any]:
+ ) -> Select[Unpack[Tuple[Any, ...]]]:
r"""Return a new :func:`_expression.select` construct with
the given entities appended to its columns clause.
"be removed in a future release. Please use "
":meth:`_expression.Select.add_columns`",
)
- def column(self, column: _ColumnsClauseArgument[Any]) -> Select[Any]:
+ def column(
+ self, column: _ColumnsClauseArgument[Any]
+ ) -> Select[Unpack[Tuple[Any, ...]]]:
"""Return a new :func:`_expression.select` construct with
the given column expression added to its columns clause.
return self.add_columns(column)
@util.preload_module("sqlalchemy.sql.util")
- def reduce_columns(self, only_synonyms: bool = True) -> Select[Any]:
+ def reduce_columns(
+ self, only_synonyms: bool = True
+ ) -> Select[Unpack[Tuple[Any, ...]]]:
"""Return a new :func:`_expression.select` construct with redundantly
named, equivalently-valued columns removed from the columns clause.
all columns that are equivalent to another are removed.
"""
- woc: Select[Any]
+ woc: Select[Unpack[Tuple[Any, ...]]]
woc = self.with_only_columns(
*util.preloaded.sql_util.reduce_columns(
self._all_selected_columns,
*entities: _ColumnsClauseArgument[Any],
maintain_column_froms: bool = False,
**__kw: Any,
- ) -> Select[Any]:
+ ) -> Select[Unpack[Tuple[Any, ...]]]:
...
@_generative
*entities: _ColumnsClauseArgument[Any],
maintain_column_froms: bool = False,
**__kw: Any,
- ) -> Select[Any]:
+ ) -> Select[Unpack[Tuple[Any, ...]]]:
r"""Return a new :func:`_expression.select` construct with its columns
clause replaced with the given entities.
meth = SelectState.get_plugin_class(self).all_selected_columns
return list(meth(self))
- def _ensure_disambiguated_names(self) -> Select[Any]:
+ def _ensure_disambiguated_names(self) -> Select[Unpack[Tuple[Any, ...]]]:
if self._label_style is LABEL_STYLE_NONE:
self = self.set_label_style(LABEL_STYLE_DISAMBIGUATE_ONLY)
return self
by this :class:`_expression.ScalarSelect`.
"""
- self.element = cast("Select[Any]", self.element).where(crit)
+ self.element = cast(
+ "Select[Unpack[Tuple[Any, ...]]]", self.element
+ ).where(crit)
return self
@overload
if TYPE_CHECKING:
- def _ungroup(self) -> Select[Any]:
+ def _ungroup(self) -> Select[Unpack[Tuple[Any, ...]]]:
...
@_generative
"""
- self.element = cast("Select[Any]", self.element).correlate(
- *fromclauses
- )
+ self.element = cast(
+ "Select[Unpack[Tuple[Any, ...]]]", self.element
+ ).correlate(*fromclauses)
return self
@_generative
"""
- self.element = cast("Select[Any]", self.element).correlate_except(
- *fromclauses
- )
+ self.element = cast(
+ "Select[Unpack[Tuple[Any, ...]]]", self.element
+ ).correlate_except(*fromclauses)
return self
"""
inherit_cache = True
- element: Union[SelectStatementGrouping[Select[Any]], ScalarSelect[Any]]
+ element: Union[
+ SelectStatementGrouping[Select[Unpack[Tuple[Any, ...]]]],
+ ScalarSelect[Any],
+ ]
def __init__(
self,
return []
def _regroup(
- self, fn: Callable[[Select[Any]], Select[Any]]
- ) -> SelectStatementGrouping[Select[Any]]:
+ self,
+ fn: Callable[
+ [Select[Unpack[Tuple[Any, ...]]]], Select[Unpack[Tuple[Any, ...]]]
+ ],
+ ) -> SelectStatementGrouping[Select[Unpack[Tuple[Any, ...]]]]:
element = self.element._ungroup()
new_element = fn(element)
assert isinstance(return_value, SelectStatementGrouping)
return return_value
- def select(self) -> Select[Any]:
+ def select(self) -> Select[Unpack[Tuple[Any, ...]]]:
r"""Return a SELECT of this :class:`_expression.Exists`.
e.g.::
result = session.execute(stmt)
- # EXPECTED_TYPE: Result[Tuple[int, str]]
+ # EXPECTED_TYPE: Result[int, str]
reveal_type(result)
result = session.execute(stmt)
- # EXPECTED_TYPE: Result[Tuple[User]]
+ # EXPECTED_TYPE: Result[User]
reveal_type(result)
result = session.execute(stmt)
- # EXPECTED_TYPE: Result[Tuple[int, str]]
+ # EXPECTED_TYPE: Result[int, str]
reveal_type(result)
result = session.execute(stmt)
- # EXPECTED_TYPE: Result[Tuple[User, User]]
+ # EXPECTED_TYPE: Result[User, User]
reveal_type(result)
@overload
def {current_fnname}(
{'self, ' if use_self else ''}{", ".join(combination)}{extra_args}
-) -> {return_type}[Tuple[{', '.join(f'_T{i}' for i in range(num_args))}]]:
+) -> {return_type}[{', '.join(f'_T{i}' for i in range(num_args))}]:
...
""", # noqa: E501
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
