--- /dev/null
+.. change::
+ :tags: bug, typing
+ :tickets: 9185
+
+ Typing improvements:
+
+ * :class:`.CursorResult` is returned for some forms of
+ :meth:`_orm.Session.execute` where DML without RETURNING is used
+ * fixed type for :paramref:`_orm.Query.with_for_update.of` parameter within
+ :meth:`_orm.Query.with_for_update`
+ * improvements to ``_DMLColumnArgument`` type used by some DML methods to
+ pass column expressions
+ * Add overload to :func:`_sql.literal` so that it is inferred that the
+ return type is ``BindParameter[NullType]`` where
+ :paramref:`_sql.literal.type_` param is None
+ * Add overloads to :meth:`_sql.ColumnElement.op` so that the inferred
+ type when :paramref:`_sql.ColumnElement.op.return_type` is not provided
+ is ``Callable[[Any], BinaryExpression[Any]]``
+ * Add missing overload to :meth:`_sql.ColumnElement.__add__`
+
+ Pull request courtesy Mehdi Gmira.
+
from .result import AsyncScalarResult
from .session import AsyncSessionTransaction
from ...engine import Connection
+ from ...engine import CursorResult
from ...engine import Engine
from ...engine import Result
from ...engine import Row
from ...orm.session import _PKIdentityArgument
from ...orm.session import _SessionBind
from ...sql.base import Executable
+ from ...sql.dml import UpdateBase
from ...sql.elements import ClauseElement
from ...sql.selectable import ForUpdateParameter
from ...sql.selectable import TypedReturnsRows
) -> Result[_T]:
...
+ @overload
+ async def execute(
+ self,
+ statement: UpdateBase,
+ 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,
+ ) -> CursorResult[Any]:
+ ...
+
@overload
async def execute(
self,
from .engine import AsyncConnection
from .engine import AsyncEngine
from ...engine import Connection
+ from ...engine import CursorResult
from ...engine import Engine
from ...engine import Result
from ...engine import Row
from ...orm.session import _SessionBindKey
from ...sql._typing import _InfoType
from ...sql.base import Executable
+ from ...sql.dml import UpdateBase
from ...sql.elements import ClauseElement
from ...sql.selectable import ForUpdateParameter
from ...sql.selectable import TypedReturnsRows
) -> Result[_T]:
...
+ @overload
+ async def execute(
+ self,
+ statement: UpdateBase,
+ 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,
+ ) -> CursorResult[Any]:
+ ...
+
@overload
async def execute(
self,
from ..sql.base import ExecutableOption
from ..sql.elements import ColumnElement
from ..sql.elements import Label
+ from ..sql.selectable import _ForUpdateOfArgument
from ..sql.selectable import _JoinTargetElement
from ..sql.selectable import _SetupJoinsElement
from ..sql.selectable import Alias
*,
nowait: bool = False,
read: bool = False,
- of: Optional[
- Union[
- _ColumnExpressionArgument[Any],
- Sequence[_ColumnExpressionArgument[Any]],
- ]
- ] = None,
+ of: Optional[_ForUpdateOfArgument] = None,
skip_locked: bool = False,
key_share: bool = False,
) -> Self:
delete_ = sql.delete(*self._raw_columns) # type: ignore
delete_._where_criteria = self._where_criteria
- result: CursorResult[Any] = cast(
- "CursorResult[Any]",
- self.session.execute(
- delete_,
- self._params,
- execution_options=self._execution_options.union(
- {"synchronize_session": synchronize_session}
- ),
+ result: CursorResult[Any] = self.session.execute(
+ delete_,
+ self._params,
+ execution_options=self._execution_options.union(
+ {"synchronize_session": synchronize_session}
),
)
bulk_del.result = result # type: ignore
upd = upd.with_dialect_options(**update_args)
upd._where_criteria = self._where_criteria
- result: CursorResult[Any] = cast(
- "CursorResult[Any]",
- self.session.execute(
- upd,
- self._params,
- execution_options=self._execution_options.union(
- {"synchronize_session": synchronize_session}
- ),
+ result: CursorResult[Any] = self.session.execute(
+ upd,
+ self._params,
+ execution_options=self._execution_options.union(
+ {"synchronize_session": synchronize_session}
),
)
bulk_ud.result = result # type: ignore
from .session import sessionmaker
from .session import SessionTransaction
from ..engine import Connection
+ from ..engine import CursorResult
from ..engine import Engine
from ..engine import Result
from ..engine import Row
from ..sql._typing import _T7
from ..sql._typing import _TypedColumnClauseArgument as _TCCA
from ..sql.base import Executable
+ from ..sql.dml import UpdateBase
from ..sql.elements import ClauseElement
from ..sql.roles import TypedColumnsClauseRole
from ..sql.selectable import ForUpdateParameter
) -> Result[_T]:
...
+ @overload
+ def execute(
+ self,
+ statement: UpdateBase,
+ 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,
+ ) -> CursorResult[Any]:
+ ...
+
@overload
def execute(
self,
from .mapper import Mapper
from .path_registry import PathRegistry
from .query import RowReturningQuery
+ from ..engine import CursorResult
from ..engine import Result
from ..engine import Row
from ..engine import RowMapping
from ..sql._typing import _TypedColumnClauseArgument as _TCCA
from ..sql.base import Executable
from ..sql.base import ExecutableOption
+ from ..sql.dml import UpdateBase
from ..sql.elements import ClauseElement
from ..sql.roles import TypedColumnsClauseRole
from ..sql.selectable import ForUpdateParameter
) -> Result[_T]:
...
+ @overload
+ def execute(
+ self,
+ statement: UpdateBase,
+ 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,
+ ) -> CursorResult[Any]:
+ ...
+
@overload
def execute(
self,
str,
_HasClauseElement,
roles.DMLColumnRole,
- "SQLCoreOperations",
+ "SQLCoreOperations[Any]",
]
"""A DML column expression. This is a "key" inside of insert().values(),
update().values(), and related.
_NMT = TypeVar("_NMT", bound="_NUMBER")
+@overload
def literal(
value: Any,
- type_: Optional[_TypeEngineArgument[_T]] = None,
+ type_: _TypeEngineArgument[_T],
literal_execute: bool = False,
) -> BindParameter[_T]:
+ ...
+
+
+@overload
+def literal(
+ value: _T,
+ type_: None = None,
+ literal_execute: bool = False,
+) -> BindParameter[_T]:
+ ...
+
+
+@overload
+def literal(
+ value: Any,
+ type_: Optional[_TypeEngineArgument[Any]] = None,
+ literal_execute: bool = False,
+) -> BindParameter[Any]:
+ ...
+
+
+def literal(
+ value: Any,
+ type_: Optional[_TypeEngineArgument[Any]] = None,
+ literal_execute: bool = False,
+) -> BindParameter[Any]:
r"""Return a literal clause, bound to a bind parameter.
Literal clauses are created automatically when non-
) -> ColumnElement[Any]:
...
+ @overload
+ def op(
+ self,
+ opstring: str,
+ precedence: int = ...,
+ is_comparison: bool = ...,
+ *,
+ return_type: _TypeEngineArgument[_OPT],
+ python_impl: Optional[Callable[..., Any]] = None,
+ ) -> Callable[[Any], BinaryExpression[_OPT]]:
+ ...
+
+ @overload
+ def op(
+ self,
+ opstring: str,
+ precedence: int = ...,
+ is_comparison: bool = ...,
+ return_type: Optional[_TypeEngineArgument[Any]] = ...,
+ python_impl: Optional[Callable[..., Any]] = ...,
+ ) -> Callable[[Any], BinaryExpression[Any]]:
+ ...
+
def op(
self,
opstring: str,
precedence: int = 0,
is_comparison: bool = False,
- return_type: Optional[_TypeEngineArgument[_OPT]] = None,
+ return_type: Optional[_TypeEngineArgument[Any]] = None,
python_impl: Optional[Callable[..., Any]] = None,
- ) -> Callable[[Any], BinaryExpression[_OPT]]:
+ ) -> Callable[[Any], BinaryExpression[Any]]:
...
def bool_op(
) -> ColumnElement[str]:
...
+ @overload
+ def __add__(self, other: Any) -> ColumnElement[Any]:
+ ...
+
def __add__(self, other: Any) -> ColumnElement[Any]:
...
reveal_type(r3)
+def t_dml_bare_insert() -> None:
+ s1 = insert(User)
+ r1 = session.execute(s1)
+ # EXPECTED_TYPE: CursorResult[Any]
+ reveal_type(r1)
+ # EXPECTED_TYPE: int
+ reveal_type(r1.rowcount)
+
+
+def t_dml_bare_update() -> None:
+ s1 = update(User)
+ r1 = session.execute(s1)
+ # EXPECTED_TYPE: CursorResult[Any]
+ reveal_type(r1)
+ # EXPECTED_TYPE: int
+ reveal_type(r1.rowcount)
+
+
+def t_dml_bare_delete() -> None:
+ s1 = delete(User)
+ r1 = session.execute(s1)
+ # EXPECTED_TYPE: CursorResult[Any]
+ reveal_type(r1)
+ # EXPECTED_TYPE: int
+ reveal_type(r1.rowcount)
+
+
def t_dml_update() -> None:
s1 = update(User).returning(User.id, User.name)
from sqlalchemy import Column
from sqlalchemy import desc
from sqlalchemy import Integer
+from sqlalchemy import literal
from sqlalchemy import MetaData
from sqlalchemy import select
from sqlalchemy import SQLColumnExpression
s9174_6 = select(user_table).with_for_update(
of=[user_table.c.id, user_table.c.email]
)
+
+# with_for_update but for query
+session = Session()
+user = session.query(User).with_for_update(of=User)
+user = session.query(User).with_for_update(of=User.id)
+user = session.query(User).with_for_update(of=[User.id, User.email])
+user = session.query(user_table).with_for_update(of=user_table)
+user = session.query(user_table).with_for_update(of=user_table.c.id)
+user = session.query(user_table).with_for_update(
+ of=[user_table.c.id, user_table.c.email]
+)
+
+# literal
+# EXPECTED_TYPE: BindParameter[str]
+reveal_type(literal("5"))
+# EXPECTED_TYPE: BindParameter[str]
+reveal_type(literal("5", None))
+# EXPECTED_TYPE: BindParameter[int]
+reveal_type(literal("123", Integer))
+# EXPECTED_TYPE: BindParameter[int]
+reveal_type(literal("123", Integer))
op_e: "ColumnElement[bool]" = col.bool_op("&")("1")
+op_a1 = col.op("&")(1)
+# EXPECTED_TYPE: BinaryExpression[Any]
+reveal_type(op_a1)
+
+
# op functions
t1 = operators.eq(A.id, 1)
select().where(t1)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
