from ..sql.type_api import TypeEngine
if typing.TYPE_CHECKING:
- from ..engine.interfaces import _IsolationLevel
+ from ..engine.interfaces import IsolationLevel
class PyODBCConnector(Connector):
def get_isolation_level_values(
self, dbapi_connection: interfaces.DBAPIConnection
- ) -> List[_IsolationLevel]:
+ ) -> List[IsolationLevel]:
return super().get_isolation_level_values(dbapi_connection) + [ # type: ignore # noqa: E501
"AUTOCOMMIT"
]
def set_isolation_level(
self,
dbapi_connection: interfaces.DBAPIConnection,
- level: _IsolationLevel,
+ level: IsolationLevel,
) -> None:
# adjust for ConnectionFairy being present
# allows attribute set e.g. "connection.autocommit = True"
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
+"""Defines :class:`_engine.Connection` and :class:`_engine.Engine`.
+
+"""
from __future__ import annotations
import contextlib
from typing import Iterator
from typing import List
from typing import Mapping
-from typing import MutableMapping
from typing import NoReturn
from typing import Optional
from typing import overload
from typing import TypeVar
from typing import Union
-from .interfaces import _IsolationLevel
from .interfaces import BindTyping
from .interfaces import ConnectionEventsTarget
from .interfaces import DBAPICursor
from .interfaces import ExceptionContext
from .interfaces import ExecuteStyle
from .interfaces import ExecutionContext
+from .interfaces import IsolationLevel
from .util import _distill_params_20
from .util import _distill_raw_params
from .util import TransactionalContext
from ..sql import compiler
from ..sql import util as sql_util
-_CompiledCacheType = MutableMapping[Any, "Compiled"]
-
if typing.TYPE_CHECKING:
from . import CursorResult
from . import ScalarResult
from .interfaces import _DBAPIAnyExecuteParams
from .interfaces import _DBAPISingleExecuteParams
from .interfaces import _ExecuteOptions
- from .interfaces import _ExecuteOptionsParameter
- from .interfaces import _SchemaTranslateMapType
+ from .interfaces import CompiledCacheType
+ from .interfaces import CoreExecuteOptionsParameter
from .interfaces import Dialect
+ from .interfaces import SchemaTranslateMapType
from .reflection import Inspector # noqa
from .url import URL
from ..event import dispatcher
from ..sql.schema import SchemaItem
from ..sql.selectable import TypedReturnsRows
-"""Defines :class:`_engine.Connection` and :class:`_engine.Engine`.
-
-"""
_T = TypeVar("_T", bound=Any)
_EMPTY_EXECUTION_OPTS: _ExecuteOptions = util.EMPTY_DICT
self.engine.logger.debug(message, *arg, **kw)
@property
- def _schema_translate_map(self) -> Optional[_SchemaTranslateMapType]:
+ def _schema_translate_map(self) -> Optional[SchemaTranslateMapType]:
return self._execution_options.get("schema_translate_map", None)
def schema_for_object(self, obj: HasSchemaAttr) -> Optional[str]:
name = obj.schema
schema_translate_map: Optional[
- Mapping[Optional[str], str]
+ SchemaTranslateMapType
] = self._execution_options.get("schema_translate_map", None)
if (
def __exit__(self, type_: Any, value: Any, traceback: Any) -> None:
self.close()
+ @overload
+ def execution_options(
+ self,
+ *,
+ compiled_cache: Optional[CompiledCacheType] = ...,
+ logging_token: str = ...,
+ isolation_level: IsolationLevel = ...,
+ no_parameters: bool = False,
+ stream_results: bool = False,
+ max_row_buffer: int = ...,
+ yield_per: int = ...,
+ insertmanyvalues_page_size: int = ...,
+ schema_translate_map: Optional[SchemaTranslateMapType] = ...,
+ **opt: Any,
+ ) -> Connection:
+ ...
+
+ @overload
+ def execution_options(self, **opt: Any) -> Connection:
+ ...
+
def execution_options(self, **opt: Any) -> Connection:
r"""Set non-SQL options for the connection which take effect
during execution.
:ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`
describing the ORM version of ``yield_per``
- :param insertmanyvalues_page_size: number of rows to format into an
+ :param insertmanyvalues_page_size: Available on: :class:`_engine.Connection`,
+ :class:`_engine.Engine`. Number of rows to format into an
INSERT statement when the statement uses "insertmanyvalues" mode,
which is a paged form of bulk insert that is used for many backends
when using :term:`executemany` execution typically in conjunction
else:
return self._dbapi_connection
- def get_isolation_level(self) -> _IsolationLevel:
+ def get_isolation_level(self) -> IsolationLevel:
"""Return the current isolation level assigned to this
:class:`_engine.Connection`.
self._handle_dbapi_exception(e, None, None, None, None)
@property
- def default_isolation_level(self) -> Optional[_IsolationLevel]:
+ def default_isolation_level(self) -> Optional[IsolationLevel]:
"""The default isolation level assigned to this
:class:`_engine.Connection`.
statement: TypedReturnsRows[Tuple[_T]],
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> Optional[_T]:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> Any:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> Any:
r"""Executes a SQL statement construct and returns a scalar object.
statement: TypedReturnsRows[Tuple[_T]],
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> ScalarResult[_T]:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> ScalarResult[Any]:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> ScalarResult[Any]:
"""Executes and returns a scalar result set, which yields scalar values
from the first column of each row.
statement: TypedReturnsRows[_T],
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> CursorResult[_T]:
...
statement: Executable,
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> CursorResult[Any]:
...
statement: Executable,
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> CursorResult[Any]:
r"""Executes a SQL statement construct and returns a
:class:`_engine.CursorResult`.
self,
func: FunctionElement[Any],
distilled_parameters: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: CoreExecuteOptionsParameter,
) -> CursorResult[Any]:
"""Execute a sql.FunctionElement object."""
self,
default: DefaultGenerator,
distilled_parameters: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: CoreExecuteOptionsParameter,
) -> Any:
"""Execute a schema.ColumnDefault object."""
self,
ddl: ExecutableDDLElement,
distilled_parameters: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: CoreExecuteOptionsParameter,
) -> CursorResult[Any]:
"""Execute a schema.DDL object."""
self,
elem: Executable,
distilled_parameters: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: CoreExecuteOptionsParameter,
) -> CursorResult[Any]:
"""Execute a sql.ClauseElement object."""
"schema_translate_map", None
)
- compiled_cache: Optional[_CompiledCacheType] = execution_options.get(
+ compiled_cache: Optional[CompiledCacheType] = execution_options.get(
"compiled_cache", self.engine._compiled_cache
)
self,
compiled: Compiled,
distilled_parameters: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter = _EMPTY_EXECUTION_OPTS,
+ execution_options: CoreExecuteOptionsParameter = _EMPTY_EXECUTION_OPTS,
) -> CursorResult[Any]:
"""Execute a sql.Compiled object.
self,
statement: str,
parameters: Optional[_DBAPIAnyExecuteParams] = None,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> CursorResult[Any]:
r"""Executes a SQL statement construct and returns a
:class:`_engine.CursorResult`.
dispatch: dispatcher[ConnectionEventsTarget]
- _compiled_cache: Optional[_CompiledCacheType]
+ _compiled_cache: Optional[CompiledCacheType]
_execution_options: _ExecuteOptions = _EMPTY_EXECUTION_OPTS
_has_events: bool = False
_sqla_logger_namespace: str = "sqlalchemy.engine.Engine"
_is_future: bool = False
- _schema_translate_map: Optional[_SchemaTranslateMapType] = None
+ _schema_translate_map: Optional[SchemaTranslateMapType] = None
_option_cls: Type[OptionEngine]
dialect: Dialect
self._execution_options = self._execution_options.union(opt)
self.dialect.set_engine_execution_options(self, opt)
+ @overload
+ def execution_options(
+ self,
+ *,
+ compiled_cache: Optional[CompiledCacheType] = ...,
+ logging_token: str = ...,
+ isolation_level: IsolationLevel = ...,
+ insertmanyvalues_page_size: int = ...,
+ schema_translate_map: Optional[SchemaTranslateMapType] = ...,
+ **opt: Any,
+ ) -> OptionEngine:
+ ...
+
+ @overload
+ def execution_options(self, **opt: Any) -> OptionEngine:
+ ...
+
def execution_options(self, **opt: Any) -> OptionEngine:
"""Return a new :class:`_engine.Engine` that will provide
:class:`_engine.Connection` objects with the given execution options.
_sa_propagate_class_events = False
dispatch: dispatcher[ConnectionEventsTarget]
- _compiled_cache: Optional[_CompiledCacheType]
+ _compiled_cache: Optional[CompiledCacheType]
dialect: Dialect
pool: Pool
url: URL
echo: log.echo_property
def __init__(
- self, proxied: Engine, execution_options: _ExecuteOptionsParameter
+ self, proxied: Engine, execution_options: CoreExecuteOptionsParameter
):
self._proxied = proxied
self.url = proxied.url
if typing.TYPE_CHECKING:
from .base import Engine
from .interfaces import _ExecuteOptions
- from .interfaces import _IsolationLevel
from .interfaces import _ParamStyle
+ from .interfaces import IsolationLevel
from .url import URL
from ..log import _EchoFlagType
from ..pool import _CreatorFnType
hide_parameters: bool = ...,
implicit_returning: Literal[True] = ...,
insertmanyvalues_page_size: int = ...,
- isolation_level: _IsolationLevel = ...,
+ isolation_level: IsolationLevel = ...,
json_deserializer: Callable[..., Any] = ...,
json_serializer: Callable[..., Any] = ...,
label_length: Optional[int] = ...,
from .interfaces import _DBAPICursorDescription
from .interfaces import _DBAPIMultiExecuteParams
from .interfaces import _ExecuteOptions
- from .interfaces import _IsolationLevel
from .interfaces import _MutableCoreSingleExecuteParams
from .interfaces import _ParamStyle
from .interfaces import DBAPIConnection
+ from .interfaces import IsolationLevel
from .row import Row
from .url import URL
from ..event import _ListenerFnType
def __init__(
self,
paramstyle: Optional[_ParamStyle] = None,
- isolation_level: Optional[_IsolationLevel] = None,
+ isolation_level: Optional[IsolationLevel] = None,
dbapi: Optional[ModuleType] = None,
implicit_returning: Literal[True] = True,
supports_native_boolean: Optional[bool] = None,
return [[], opts]
def set_engine_execution_options(
- self, engine: Engine, opts: Mapping[str, str]
+ self, engine: Engine, opts: Mapping[str, Any]
) -> None:
supported_names = set(self.connection_characteristics).intersection(
opts
)
if supported_names:
- characteristics: Mapping[str, str] = util.immutabledict(
+ characteristics: Mapping[str, Any] = util.immutabledict(
(name, opts[name]) for name in supported_names
)
)
def set_connection_execution_options(
- self, connection: Connection, opts: Mapping[str, str]
+ self, connection: Connection, opts: Mapping[str, Any]
) -> None:
supported_names = set(self.connection_characteristics).intersection(
opts
)
if supported_names:
- characteristics: Mapping[str, str] = util.immutabledict(
+ characteristics: Mapping[str, Any] = util.immutabledict(
(name, opts[name]) for name in supported_names
)
self._set_connection_characteristics(connection, characteristics)
_AnyMultiExecuteParams = _DBAPIMultiExecuteParams
_AnyExecuteParams = _DBAPIAnyExecuteParams
-
-_ExecuteOptions = immutabledict[str, Any]
-_ExecuteOptionsParameter = Mapping[str, Any]
-_SchemaTranslateMapType = Mapping[str, str]
+CompiledCacheType = MutableMapping[Any, "Compiled"]
+SchemaTranslateMapType = Mapping[Optional[str], Optional[str]]
_ImmutableExecuteOptions = immutabledict[str, Any]
_GenericSetInputSizesType = List[Tuple[str, Any, "TypeEngine[Any]"]]
-_IsolationLevel = Literal[
+IsolationLevel = Literal[
"SERIALIZABLE",
"REPEATABLE READ",
"READ COMMITTED",
]
+class _CoreKnownExecutionOptions(TypedDict, total=False):
+ compiled_cache: Optional[CompiledCacheType]
+ logging_token: str
+ isolation_level: IsolationLevel
+ no_parameters: bool
+ stream_results: bool
+ max_row_buffer: int
+ yield_per: int
+ insertmanyvalues_page_size: int
+ schema_translate_map: Optional[SchemaTranslateMapType]
+
+
+_ExecuteOptions = immutabledict[str, Any]
+CoreExecuteOptionsParameter = Union[
+ _CoreKnownExecutionOptions, Mapping[str, Any]
+]
+
+
class ReflectedIdentity(TypedDict):
"""represent the reflected IDENTITY structure of a column, corresponding
to the :class:`_schema.Identity` construct.
# NOTE: this does not take into effect engine-level isolation level.
# not clear if this should be changed, seems like it should
- default_isolation_level: Optional[_IsolationLevel]
+ default_isolation_level: Optional[IsolationLevel]
"""the isolation that is implicitly present on new connections"""
# create_engine() -> isolation_level currently goes here
- _on_connect_isolation_level: Optional[_IsolationLevel]
+ _on_connect_isolation_level: Optional[IsolationLevel]
execution_ctx_cls: Type["ExecutionContext"]
"""a :class:`.ExecutionContext` class used to handle statement execution"""
raise NotImplementedError()
def set_isolation_level(
- self, dbapi_connection: DBAPIConnection, level: _IsolationLevel
+ self, dbapi_connection: DBAPIConnection, level: IsolationLevel
) -> None:
"""Given a DBAPI connection, set its isolation level.
def get_isolation_level(
self, dbapi_connection: DBAPIConnection
- ) -> _IsolationLevel:
+ ) -> IsolationLevel:
"""Given a DBAPI connection, return its isolation level.
When working with a :class:`_engine.Connection` object,
def get_default_isolation_level(
self, dbapi_conn: DBAPIConnection
- ) -> _IsolationLevel:
+ ) -> IsolationLevel:
"""Given a DBAPI connection, return its isolation level, or
a default isolation level if one cannot be retrieved.
def get_isolation_level_values(
self, dbapi_conn: DBAPIConnection
- ) -> List[_IsolationLevel]:
+ ) -> List[IsolationLevel]:
"""return a sequence of string isolation level names that are accepted
by this dialect.
raise NotImplementedError()
def _assert_and_set_isolation_level(
- self, dbapi_conn: DBAPIConnection, level: _IsolationLevel
+ self, dbapi_conn: DBAPIConnection, level: IsolationLevel
) -> None:
raise NotImplementedError()
raise NotImplementedError()
def set_engine_execution_options(
- self, engine: Engine, opt: _ExecuteOptionsParameter
+ self, engine: Engine, opts: CoreExecuteOptionsParameter
) -> None:
"""Establish execution options for a given engine.
raise NotImplementedError()
def set_connection_execution_options(
- self, connection: Connection, opt: _ExecuteOptionsParameter
+ self, connection: Connection, opts: CoreExecuteOptionsParameter
) -> None:
"""Establish execution options for a given connection.
if typing.TYPE_CHECKING:
from .base import Engine
from .interfaces import _CoreAnyExecuteParams
- from .interfaces import _ExecuteOptionsParameter
+ from .interfaces import CoreExecuteOptionsParameter
from .interfaces import Dialect
from .url import URL
from ..sql.base import Executable
self,
obj: Executable,
parameters: Optional[_CoreAnyExecuteParams] = None,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> Any:
return self._execute_impl(obj, parameters)
from ...engine.interfaces import _CoreSingleExecuteParams
from ...engine.interfaces import _DBAPIAnyExecuteParams
from ...engine.interfaces import _ExecuteOptions
- from ...engine.interfaces import _ExecuteOptionsParameter
- from ...engine.interfaces import _IsolationLevel
+ from ...engine.interfaces import CompiledCacheType
+ from ...engine.interfaces import CoreExecuteOptionsParameter
from ...engine.interfaces import Dialect
+ from ...engine.interfaces import IsolationLevel
+ from ...engine.interfaces import SchemaTranslateMapType
from ...engine.result import ScalarResult
from ...engine.url import URL
from ...pool import Pool
self._proxied.invalidate, exception=exception
)
- async def get_isolation_level(self) -> _IsolationLevel:
+ async def get_isolation_level(self) -> IsolationLevel:
return await greenlet_spawn(self._proxied.get_isolation_level)
def in_transaction(self) -> bool:
else:
return None
+ @overload
+ async def execution_options(
+ self,
+ *,
+ compiled_cache: Optional[CompiledCacheType] = ...,
+ logging_token: str = ...,
+ isolation_level: IsolationLevel = ...,
+ no_parameters: bool = False,
+ stream_results: bool = False,
+ max_row_buffer: int = ...,
+ yield_per: int = ...,
+ insertmanyvalues_page_size: int = ...,
+ schema_translate_map: Optional[SchemaTranslateMapType] = ...,
+ **opt: Any,
+ ) -> AsyncConnection:
+ ...
+
+ @overload
+ async def execution_options(self, **opt: Any) -> AsyncConnection:
+ ...
+
async def execution_options(self, **opt: Any) -> AsyncConnection:
r"""Set non-SQL options for the connection which take effect
during execution.
self,
statement: str,
parameters: Optional[_DBAPIAnyExecuteParams] = None,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> CursorResult[Any]:
r"""Executes a driver-level SQL string and return buffered
:class:`_engine.Result`.
statement: TypedReturnsRows[_T],
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> AsyncResult[_T]:
...
statement: Executable,
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> AsyncResult[Any]:
...
statement: Executable,
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> AsyncResult[Any]:
"""Execute a statement and return a streaming
:class:`_asyncio.AsyncResult` object."""
statement: TypedReturnsRows[_T],
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> CursorResult[_T]:
...
statement: Executable,
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> CursorResult[Any]:
...
statement: Executable,
parameters: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> CursorResult[Any]:
r"""Executes a SQL statement construct and return a buffered
:class:`_engine.Result`.
statement: TypedReturnsRows[Tuple[_T]],
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> Optional[_T]:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> Any:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> Any:
r"""Executes a SQL statement construct and returns a scalar object.
statement: TypedReturnsRows[Tuple[_T]],
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> ScalarResult[_T]:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> ScalarResult[Any]:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> ScalarResult[Any]:
r"""Executes a SQL statement construct and returns a scalar objects.
statement: TypedReturnsRows[Tuple[_T]],
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> AsyncScalarResult[_T]:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> AsyncScalarResult[Any]:
...
statement: Executable,
parameters: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[CoreExecuteOptionsParameter] = None,
) -> AsyncScalarResult[Any]:
r"""Executes a SQL statement and returns a streaming scalar result
object.
"""
return await greenlet_spawn(self.sync_engine.raw_connection)
+ @overload
+ def execution_options(
+ self,
+ *,
+ compiled_cache: Optional[CompiledCacheType] = ...,
+ logging_token: str = ...,
+ isolation_level: IsolationLevel = ...,
+ insertmanyvalues_page_size: int = ...,
+ schema_translate_map: Optional[SchemaTranslateMapType] = ...,
+ **opt: Any,
+ ) -> AsyncEngine:
+ ...
+
+ @overload
+ def execution_options(self, **opt: Any) -> AsyncEngine:
+ ...
+
def execution_options(self, **opt: Any) -> AsyncEngine:
"""Return a new :class:`_asyncio.AsyncEngine` that will provide
:class:`_asyncio.AsyncConnection` objects with the given execution
from ...engine import RowMapping
from ...engine.interfaces import _CoreAnyExecuteParams
from ...engine.interfaces import _CoreSingleExecuteParams
- from ...engine.interfaces import _ExecuteOptionsParameter
from ...engine.result import ScalarResult
from ...orm._typing import _IdentityKeyType
from ...orm._typing import _O
+ from ...orm._typing import OrmExecuteOptionsParameter
from ...orm.interfaces import ORMOption
from ...orm.session import _BindArguments
from ...orm.session import _EntityBindKey
populate_existing: bool = False,
with_for_update: Optional[ForUpdateArg] = None,
identity_token: Optional[Any] = None,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
) -> Optional[_O]:
r"""Return an instance based on the given primary key identifier,
or ``None`` if not found.
statement: TypedReturnsRows[_T],
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Result[Any]:
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Optional[_T]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Any:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Any:
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[_T]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[Any]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[Any]:
statement: TypedReturnsRows[_T],
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncResult[_T]:
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncResult[Any]:
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncResult[Any]:
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncScalarResult[_T]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncScalarResult[Any]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncScalarResult[Any]:
from ...engine import ScalarResult
from ...engine.interfaces import _CoreAnyExecuteParams
from ...engine.interfaces import _CoreSingleExecuteParams
- from ...engine.interfaces import _ExecuteOptionsParameter
from ...event import dispatcher
from ...orm._typing import _IdentityKeyType
from ...orm._typing import _O
+ from ...orm._typing import OrmExecuteOptionsParameter
from ...orm.identity import IdentityMap
from ...orm.interfaces import ORMOption
from ...orm.session import _BindArguments
statement: TypedReturnsRows[_T],
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Result[Any]:
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Optional[_T]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Any:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Any:
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[_T]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[Any]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[Any]:
populate_existing: bool = False,
with_for_update: Optional[ForUpdateArg] = None,
identity_token: Optional[Any] = None,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
) -> Optional[_O]:
"""Return an instance based on the given primary key identifier,
statement: TypedReturnsRows[_T],
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncResult[_T]:
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncResult[Any]:
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncResult[Any]:
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncScalarResult[_T]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncScalarResult[Any]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> AsyncScalarResult[Any]:
+# orm/_typing.py
+# Copyright (C) 2022 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
+
from __future__ import annotations
import operator
from typing import TypeVar
from typing import Union
+from ..engine.interfaces import _CoreKnownExecutionOptions
from ..sql import roles
+from ..sql._orm_types import DMLStrategyArgument as DMLStrategyArgument
+from ..sql._orm_types import (
+ SynchronizeSessionArgument as SynchronizeSessionArgument,
+)
from ..sql._typing import _HasClauseElement
from ..sql.elements import ColumnElement
from ..util.typing import Protocol
_ORMCOLEXPR = TypeVar("_ORMCOLEXPR", bound=ColumnElement[Any])
+class _OrmKnownExecutionOptions(_CoreKnownExecutionOptions, total=False):
+ populate_existing: bool
+ autoflush: bool
+ synchronize_session: SynchronizeSessionArgument
+ dml_strategy: DMLStrategyArgument
+ is_delete_using: bool
+ is_update_from: bool
+
+
+OrmExecuteOptionsParameter = Union[
+ _OrmKnownExecutionOptions, Mapping[str, Any]
+]
+
+
class _ORMAdapterProto(Protocol):
"""protocol for the :class:`.AliasedInsp._orm_adapt_element` method
which is a synonym for :class:`.AliasedInsp._adapt_element`.
from ..util.typing import Literal
if TYPE_CHECKING:
+ from ._typing import DMLStrategyArgument
+ from ._typing import OrmExecuteOptionsParameter
+ from ._typing import SynchronizeSessionArgument
from .mapper import Mapper
from .session import _BindArguments
from .session import ORMExecuteState
from ..engine import Connection
from ..engine import cursor
from ..engine.interfaces import _CoreAnyExecuteParams
- from ..engine.interfaces import _ExecuteOptionsParameter
_O = TypeVar("_O", bound=object)
-_SynchronizeSessionArgument = Literal[False, "auto", "evaluate", "fetch"]
-_DMLStrategyArgument = Literal["bulk", "raw", "orm", "auto"]
-
-
@overload
def _bulk_insert(
mapper: Mapper[_O],
return_defaults: bool,
render_nulls: bool,
use_orm_insert_stmt: Literal[None] = ...,
- execution_options: Optional[_ExecuteOptionsParameter] = ...,
+ execution_options: Optional[OrmExecuteOptionsParameter] = ...,
) -> None:
...
return_defaults: bool,
render_nulls: bool,
use_orm_insert_stmt: Optional[dml.Insert] = ...,
- execution_options: Optional[_ExecuteOptionsParameter] = ...,
+ execution_options: Optional[OrmExecuteOptionsParameter] = ...,
) -> cursor.CursorResult[Any]:
...
return_defaults: bool,
render_nulls: bool,
use_orm_insert_stmt: Optional[dml.Insert] = None,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[OrmExecuteOptionsParameter] = None,
) -> Optional[cursor.CursorResult[Any]]:
base_mapper = mapper.base_mapper
class BulkUDCompileState(ORMDMLState):
class default_update_options(Options):
- _dml_strategy: _DMLStrategyArgument = "auto"
- _synchronize_session: _SynchronizeSessionArgument = "auto"
+ _dml_strategy: DMLStrategyArgument = "auto"
+ _synchronize_session: SynchronizeSessionArgument = "auto"
_can_use_returning: bool = False
_is_delete_using: bool = False
_is_update_from: bool = False
@CompileState.plugin_for("orm", "insert")
class BulkORMInsert(ORMDMLState, InsertDMLState):
class default_insert_options(Options):
- _dml_strategy: _DMLStrategyArgument = "auto"
+ _dml_strategy: DMLStrategyArgument = "auto"
_render_nulls: bool = False
_return_defaults: bool = False
_subject_mapper: Optional[Mapper[Any]] = None
session: Session,
statement: dml.Insert,
params: _CoreAnyExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: OrmExecuteOptionsParameter,
bind_arguments: _BindArguments,
conn: Connection,
) -> _result.Result:
session: Session,
statement: dml.Update,
params: _CoreAnyExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: OrmExecuteOptionsParameter,
bind_arguments: _BindArguments,
conn: Connection,
) -> _result.Result:
session: Session,
statement: dml.Delete,
params: _CoreAnyExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: OrmExecuteOptionsParameter,
bind_arguments: _BindArguments,
conn: Connection,
) -> _result.Result:
if TYPE_CHECKING:
from ._typing import _InternalEntityType
+ from ._typing import OrmExecuteOptionsParameter
from .loading import PostLoad
from .mapper import Mapper
from .query import Query
from .session import Session
from ..engine import Result
from ..engine.interfaces import _CoreSingleExecuteParams
- from ..engine.interfaces import _ExecuteOptionsParameter
from ..sql._typing import _ColumnsClauseArgument
from ..sql.compiler import SQLCompiler
from ..sql.dml import _DMLTableElement
Type[QueryContext.default_load_options],
QueryContext.default_load_options,
],
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[OrmExecuteOptionsParameter] = None,
bind_arguments: Optional[_BindArguments] = None,
):
self.load_options = load_options
from ._typing import _EntityType
from ._typing import _ExternalEntityType
from ._typing import _InternalEntityType
- from .bulk_persistence import _SynchronizeSessionArgument
+ from ._typing import SynchronizeSessionArgument
from .mapper import Mapper
from .path_registry import PathRegistry
from .session import _PKIdentityArgument
from .state import InstanceState
from ..engine.cursor import CursorResult
from ..engine.interfaces import _ImmutableExecuteOptions
+ from ..engine.interfaces import CompiledCacheType
+ from ..engine.interfaces import IsolationLevel
+ from ..engine.interfaces import SchemaTranslateMapType
from ..engine.result import FrozenResult
from ..engine.result import ScalarResult
from ..sql._typing import _ColumnExpressionArgument
"""
return self._execution_options
+ @overload
+ def execution_options(
+ self: SelfQuery,
+ *,
+ compiled_cache: Optional[CompiledCacheType] = ...,
+ logging_token: str = ...,
+ isolation_level: IsolationLevel = ...,
+ no_parameters: bool = False,
+ stream_results: bool = False,
+ max_row_buffer: int = ...,
+ yield_per: int = ...,
+ insertmanyvalues_page_size: int = ...,
+ schema_translate_map: Optional[SchemaTranslateMapType] = ...,
+ populate_existing: bool = False,
+ autoflush: bool = False,
+ **opt: Any,
+ ) -> SelfQuery:
+ ...
+
+ @overload
+ def execution_options(self: SelfQuery, **opt: Any) -> SelfQuery:
+ ...
+
@_generative
def execution_options(self: SelfQuery, **kwargs: Any) -> SelfQuery:
"""Set non-SQL options which take effect during execution.
)
def delete(
- self, synchronize_session: _SynchronizeSessionArgument = "auto"
+ self, synchronize_session: SynchronizeSessionArgument = "auto"
) -> int:
r"""Perform a DELETE with an arbitrary WHERE clause.
def update(
self,
values: Dict[_DMLColumnArgument, Any],
- synchronize_session: _SynchronizeSessionArgument = "auto",
+ synchronize_session: SynchronizeSessionArgument = "auto",
update_args: Optional[Dict[Any, Any]] = None,
) -> int:
r"""Perform an UPDATE with an arbitrary WHERE clause.
if TYPE_CHECKING:
from ._typing import _EntityType
from ._typing import _IdentityKeyType
+ from ._typing import OrmExecuteOptionsParameter
from .identity import IdentityMap
from .interfaces import ORMOption
from .mapper import Mapper
from ..engine.interfaces import _CoreAnyExecuteParams
from ..engine.interfaces import _CoreSingleExecuteParams
from ..engine.interfaces import _ExecuteOptions
- from ..engine.interfaces import _ExecuteOptionsParameter
from ..engine.result import ScalarResult
from ..sql._typing import _ColumnsClauseArgument
from ..sql._typing import _T0
statement: TypedReturnsRows[_T],
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
populate_existing: bool = False,
with_for_update: Optional[ForUpdateArg] = None,
identity_token: Optional[Any] = None,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
) -> Optional[_O]:
r"""Return an instance based on the given primary key identifier,
or ``None`` if not found.
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Optional[_T]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Any:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Any:
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[_T]:
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[Any]:
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[Any]:
from ._typing import _EntityType
from ._typing import _IdentityKeyType
from ._typing import _InstanceDict
+ from ._typing import OrmExecuteOptionsParameter
from .interfaces import ORMOption
from .interfaces import UserDefinedOption
from .mapper import Mapper
from ..engine.interfaces import _CoreAnyExecuteParams
from ..engine.interfaces import _CoreSingleExecuteParams
from ..engine.interfaces import _ExecuteOptions
- from ..engine.interfaces import _ExecuteOptionsParameter
from ..engine.result import ScalarResult
from ..event import _InstanceLevelDispatch
from ..sql._typing import _ColumnsClauseArgument
self,
statement: Optional[Executable] = None,
params: Optional[_CoreAnyExecuteParams] = None,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[OrmExecuteOptionsParameter] = None,
bind_arguments: Optional[_BindArguments] = None,
) -> Result[Any]:
"""Execute the statement represented by this
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: TypedReturnsRows[_T],
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
_parent_execute_state: Optional[Any] = None,
_add_event: Optional[Any] = None,
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Optional[_T]:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Any:
statement: Executable,
params: Optional[_CoreSingleExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> Any:
statement: TypedReturnsRows[Tuple[_T]],
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[_T]:
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[Any]:
statement: Executable,
params: Optional[_CoreAnyExecuteParams] = None,
*,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
bind_arguments: Optional[_BindArguments] = None,
**kw: Any,
) -> ScalarResult[Any]:
populate_existing: bool = False,
with_for_update: Optional[ForUpdateArg] = None,
identity_token: Optional[Any] = None,
- execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
+ execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT,
) -> Optional[_O]:
"""Return an instance based on the given primary key identifier,
or ``None`` if not found.
populate_existing: bool = False,
with_for_update: Optional[ForUpdateArg] = None,
identity_token: Optional[Any] = None,
- execution_options: Optional[_ExecuteOptionsParameter] = None,
+ execution_options: Optional[OrmExecuteOptionsParameter] = None,
) -> Optional[_O]:
# convert composite types to individual args
--- /dev/null
+# sql/_orm_types.py
+# Copyright (C) 2022 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
+
+"""ORM types that need to present specifically for **documentation only** of
+the Executable.execution_options() method, which includes options that
+are meaningful to the ORM.
+
+"""
+
+
+from __future__ import annotations
+
+from ..util.typing import Literal
+
+SynchronizeSessionArgument = Literal[False, "auto", "evaluate", "fetch"]
+DMLStrategyArgument = Literal["bulk", "raw", "orm", "auto"]
+# sql/_typing.py
+# Copyright (C) 2022 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
+
from __future__ import annotations
import operator
from . import coercions
from . import elements
from . import type_api
+ from ._orm_types import DMLStrategyArgument
+ from ._orm_types import SynchronizeSessionArgument
from .elements import BindParameter
from .elements import ClauseList
from .elements import ColumnClause # noqa
from .selectable import FromClause
from ..engine import Connection
from ..engine import CursorResult
- from ..engine.base import _CompiledCacheType
from ..engine.interfaces import _CoreMultiExecuteParams
from ..engine.interfaces import _ExecuteOptions
- from ..engine.interfaces import _ExecuteOptionsParameter
from ..engine.interfaces import _ImmutableExecuteOptions
- from ..engine.interfaces import _SchemaTranslateMapType
from ..engine.interfaces import CacheStats
from ..engine.interfaces import Compiled
+ from ..engine.interfaces import CompiledCacheType
+ from ..engine.interfaces import CoreExecuteOptionsParameter
from ..engine.interfaces import Dialect
+ from ..engine.interfaces import IsolationLevel
+ from ..engine.interfaces import SchemaTranslateMapType
from ..event import dispatcher
if not TYPE_CHECKING:
self,
dialect: Dialect,
*,
- compiled_cache: Optional[_CompiledCacheType],
+ compiled_cache: Optional[CompiledCacheType],
column_keys: List[str],
for_executemany: bool = False,
- schema_translate_map: Optional[_SchemaTranslateMapType] = None,
+ schema_translate_map: Optional[SchemaTranslateMapType] = None,
**kw: Any,
) -> Tuple[
Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
self,
connection: Connection,
distilled_params: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: CoreExecuteOptionsParameter,
) -> CursorResult[Any]:
...
self,
connection: Connection,
distilled_params: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: CoreExecuteOptionsParameter,
) -> Any:
...
self._with_context_options += ((callable_, cache_args),)
return self
+ @overload
+ def execution_options(
+ self: SelfExecutable,
+ *,
+ compiled_cache: Optional[CompiledCacheType] = ...,
+ logging_token: str = ...,
+ isolation_level: IsolationLevel = ...,
+ no_parameters: bool = False,
+ stream_results: bool = False,
+ max_row_buffer: int = ...,
+ yield_per: int = ...,
+ insertmanyvalues_page_size: int = ...,
+ schema_translate_map: Optional[SchemaTranslateMapType] = ...,
+ populate_existing: bool = False,
+ autoflush: bool = False,
+ synchronize_session: SynchronizeSessionArgument = ...,
+ dml_strategy: DMLStrategyArgument = ...,
+ is_delete_using: bool = ...,
+ is_update_from: bool = ...,
+ **opt: Any,
+ ) -> SelfExecutable:
+ ...
+
+ @overload
+ def execution_options(self: SelfExecutable, **opt: Any) -> SelfExecutable:
+ ...
+
@_generative
def execution_options(self: SelfExecutable, **kw: Any) -> SelfExecutable:
"""Set non-SQL options for the statement which take effect during
from ..engine.interfaces import _CoreSingleExecuteParams
from ..engine.interfaces import _ExecuteOptions
from ..engine.interfaces import _MutableCoreSingleExecuteParams
- from ..engine.interfaces import _SchemaTranslateMapType
from ..engine.interfaces import Dialect
+ from ..engine.interfaces import SchemaTranslateMapType
_FromHintsType = Dict["FromClause", str]
_result_columns: Optional[List[ResultColumnsEntry]] = None
- schema_translate_map: Optional[_SchemaTranslateMapType] = None
+ schema_translate_map: Optional[SchemaTranslateMapType] = None
execution_options: _ExecuteOptions = util.EMPTY_DICT
"""
self,
dialect: Dialect,
statement: Optional[ClauseElement],
- schema_translate_map: Optional[_SchemaTranslateMapType] = None,
+ schema_translate_map: Optional[SchemaTranslateMapType] = None,
render_schema_translate: bool = False,
compile_kwargs: Mapping[str, Any] = util.immutabledict(),
):
self,
dialect: Dialect,
statement: ExecutableDDLElement,
- schema_translate_map: Optional[_SchemaTranslateMapType] = ...,
+ schema_translate_map: Optional[SchemaTranslateMapType] = ...,
render_schema_translate: bool = ...,
compile_kwargs: Mapping[str, Any] = ...,
):
from .schema import SchemaItem
from .schema import Sequence
from .schema import Table
- from ..engine.base import _CompiledCacheType
from ..engine.base import Connection
- from ..engine.interfaces import _SchemaTranslateMapType
from ..engine.interfaces import CacheStats
+ from ..engine.interfaces import CompiledCacheType
from ..engine.interfaces import Dialect
+ from ..engine.interfaces import SchemaTranslateMapType
class BaseDDLElement(ClauseElement):
self,
dialect: Dialect,
*,
- compiled_cache: Optional[_CompiledCacheType],
+ compiled_cache: Optional[CompiledCacheType],
column_keys: List[str],
for_executemany: bool = False,
- schema_translate_map: Optional[_SchemaTranslateMapType] = None,
+ schema_translate_map: Optional[SchemaTranslateMapType] = None,
**kw: Any,
) -> Tuple[
Compiled, Optional[typing_Sequence[BindParameter[Any]]], CacheStats
from ..engine import Connection
from ..engine import Dialect
from ..engine import Engine
- from ..engine.base import _CompiledCacheType
from ..engine.interfaces import _CoreMultiExecuteParams
from ..engine.interfaces import _ExecuteOptions
- from ..engine.interfaces import _SchemaTranslateMapType
from ..engine.interfaces import CacheStats
+ from ..engine.interfaces import CompiledCacheType
+ from ..engine.interfaces import SchemaTranslateMapType
from ..engine.result import Result
_NUMERIC = Union[float, Decimal]
self,
dialect: Dialect,
*,
- compiled_cache: Optional[_CompiledCacheType],
+ compiled_cache: Optional[CompiledCacheType],
column_keys: List[str],
for_executemany: bool = False,
- schema_translate_map: Optional[_SchemaTranslateMapType] = None,
+ schema_translate_map: Optional[SchemaTranslateMapType] = None,
**kw: Any,
) -> typing_Tuple[
Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
from ..engine.base import Connection
from ..engine.cursor import CursorResult
from ..engine.interfaces import _CoreMultiExecuteParams
- from ..engine.interfaces import _ExecuteOptionsParameter
+ from ..engine.interfaces import CoreExecuteOptionsParameter
_T = TypeVar("_T", bound=Any)
self,
connection: Connection,
distilled_params: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: CoreExecuteOptionsParameter,
) -> CursorResult[Any]:
return connection._execute_function(
self, distilled_params, execution_options
from ..engine import Connection
from ..engine import Engine
from ..engine.interfaces import _CoreMultiExecuteParams
- from ..engine.interfaces import _ExecuteOptionsParameter
+ from ..engine.interfaces import CoreExecuteOptionsParameter
from ..engine.interfaces import ExecutionContext
from ..engine.mock import MockConnection
from ..engine.reflection import _ReflectionInfo
self,
connection: Connection,
distilled_params: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: CoreExecuteOptionsParameter,
) -> Any:
util.warn_deprecated(
"Using the .execute() method to invoke a "
self,
connection: Connection,
distilled_params: _CoreMultiExecuteParams,
- execution_options: _ExecuteOptionsParameter,
+ execution_options: CoreExecuteOptionsParameter,
) -> Any:
return connection._execute_default(
self, distilled_params, execution_options
_T = TypeVar("_T", bound=Any)
if TYPE_CHECKING:
- import sqlalchemy
from ._typing import _ColumnExpressionArgument
from ._typing import _FromClauseArgument
from ._typing import _JoinTargetArgument
from .cache_key import _CacheKeyTraversalType
from .compiler import SQLCompiler
from .dml import Delete
+ from .dml import Insert
from .dml import Update
from .elements import KeyedColumnElement
from .elements import Label
c.table = self
@util.preload_module("sqlalchemy.sql.dml")
- def insert(self) -> sqlalchemy.sql.expression.Insert:
+ def insert(self) -> Insert:
"""Generate an :func:`_expression.insert` construct against this
:class:`_expression.TableClause`.
lambda: util.py310, "Python 3.10 or above required"
)
+ @property
+ def python311(self):
+ return exclusions.only_if(
+ lambda: util.py311, "Python 3.11 or above required"
+ )
+
@property
def cpython(self):
return exclusions.only_if(
install_requires =
importlib-metadata;python_version<"3.8"
greenlet != 0.4.17;(platform_machine=='aarch64' or (platform_machine=='ppc64le' or (platform_machine=='x86_64' or (platform_machine=='amd64' or (platform_machine=='AMD64' or (platform_machine=='win32' or platform_machine=='WIN32'))))))
- typing-extensions >= 4.1.0
+ typing-extensions >= 4.2.0
[options.extras_require]
asyncio =
args.append(path)
+ # I set this locally but for the suite here needs to be
+ # disabled
+ os.environ.pop("MYPY_FORCE_COLOR", None)
+
result = api.run(args)
return result
--- /dev/null
+from sqlalchemy import testing
+from sqlalchemy.engine.base import Connection
+from sqlalchemy.engine.base import Engine
+from sqlalchemy.engine.interfaces import CoreExecuteOptionsParameter
+from sqlalchemy.ext.asyncio.engine import AsyncConnection
+from sqlalchemy.ext.asyncio.engine import AsyncEngine
+from sqlalchemy.orm._typing import OrmExecuteOptionsParameter
+from sqlalchemy.orm.query import Query
+from sqlalchemy.sql.base import Executable
+from sqlalchemy.testing import fixtures
+from sqlalchemy.testing.assertions import eq_
+
+engine_execution_options = {
+ "compiled_cache": "Optional[CompiledCacheType]",
+ "logging_token": "str",
+ "isolation_level": "IsolationLevel",
+ "insertmanyvalues_page_size": "int",
+ "schema_translate_map": "Optional[SchemaTranslateMapType]",
+ "opt": "Any",
+}
+core_execution_options = {
+ **engine_execution_options,
+ "no_parameters": "bool",
+ "stream_results": "bool",
+ "max_row_buffer": "int",
+ "yield_per": "int",
+}
+
+orm_dql_execution_options = {
+ **core_execution_options,
+ "populate_existing": "bool",
+ "autoflush": "bool",
+}
+
+orm_dml_execution_options = {
+ "synchronize_session": "SynchronizeSessionArgument",
+ "dml_strategy": "DMLStrategyArgument",
+ "is_delete_using": "bool",
+ "is_update_from": "bool",
+}
+
+orm_execution_options = {
+ **orm_dql_execution_options,
+ **orm_dml_execution_options,
+}
+
+
+class OverloadTest(fixtures.TestBase):
+ # NOTE: get_overloads is python 3.11. typing_extensions implements it
+ # but for it to work the typing_extensions overload needs to be use and
+ # it can only be imported directly from typing_extensions in all modules
+ # that use it otherwise flake8 (pyflakes actually) will flag it with F811
+ __requires__ = ("python311",)
+
+ @testing.combinations(
+ (Engine, engine_execution_options),
+ (Connection, core_execution_options),
+ (AsyncEngine, engine_execution_options),
+ (AsyncConnection, core_execution_options),
+ (Query, orm_dql_execution_options),
+ (Executable, orm_execution_options),
+ )
+ def test_methods(self, class_, expected):
+ from typing import get_overloads
+
+ overloads = get_overloads(getattr(class_, "execution_options"))
+ eq_(len(overloads), 2)
+ annotations = overloads[0].__annotations__.copy()
+ annotations.pop("self", None)
+ annotations.pop("return", None)
+ eq_(annotations, expected)
+ annotations = overloads[1].__annotations__.copy()
+ annotations.pop("self", None)
+ annotations.pop("return", None)
+ eq_(annotations, {"opt": "Any"})
+
+ @testing.combinations(
+ (CoreExecuteOptionsParameter, core_execution_options),
+ (OrmExecuteOptionsParameter, orm_execution_options),
+ )
+ def test_typed_dicts(self, typ, expected):
+ # we currently expect these to be union types with first entry
+ # is the typed dict
+
+ typed_dict = typ.__args__[0]
+
+ expected = dict(expected)
+ expected.pop("opt")
+
+ assert_annotations = {
+ key: fwd_ref.__forward_arg__
+ for key, fwd_ref in typed_dict.__annotations__.items()
+ }
+ eq_(assert_annotations, expected)
current_ov[:] = []
break
- fn_match = re.match(r"^ (?: )?(?:async )?def (.*)\($", line)
+ fn_match = re.match(r"^ (?: )?(?:async )?def (.*)\(", line)
if fn_match and fn_match.group(1) != fn.__name__:
current_ov[:] = []
break
if re.match(r"^ if (?:typing\.)?TYPE_CHECKING:", line):
output.append(line)
+ current_ov[:] = []
output.reverse()
return output
# this is to satisfy the mypy plugin dependency
# when slotscheck imports sqlalchemy.mypy modules
sqlalchemy[mypy]
+ # required by generate_tuple_map_overloads
+ zimports
allowlist_externals =
env
+ git
+ sh
commands =
flake8 ./lib/ ./test/ ./examples/ setup.py doc/build/conf.py {posargs}
black --check ./lib/ ./test/ ./examples/ setup.py doc/build/conf.py
slotscheck -m sqlalchemy
env DISABLE_SQLALCHEMY_CEXT_RUNTIME=1 slotscheck -m sqlalchemy
python ./tools/format_docs_code.py --check
+ sh -c 'python ./tools/generate_tuple_map_overloads.py && git diff --exit-code'
+ sh -c 'python ./tools/generate_proxy_methods.py && git diff --exit-code'
# "pep8" env was renamed to "lint".
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
