Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# engine/default.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9"""Default implementations of per-dialect sqlalchemy.engine classes.
11These are semi-private implementation classes which are only of importance
12to database dialect authors; dialects will usually use the classes here
13as the base class for their own corresponding classes.
15"""
17from __future__ import annotations
19import functools
20import operator
21import random
22import re
23from time import perf_counter
24import typing
25from typing import Any
26from typing import Callable
27from typing import cast
28from typing import Dict
29from typing import Final
30from typing import Iterable
31from typing import List
32from typing import Literal
33from typing import Mapping
34from typing import MutableMapping
35from typing import MutableSequence
36from typing import Optional
37from typing import Sequence
38from typing import Set
39from typing import Tuple
40from typing import Type
41from typing import TYPE_CHECKING
42from typing import Union
43import weakref
45from . import characteristics
46from . import cursor as _cursor
47from . import interfaces
48from . import reflection
49from .base import Connection
50from .interfaces import CacheStats
51from .interfaces import DBAPICursor
52from .interfaces import Dialect
53from .interfaces import ExecuteStyle
54from .interfaces import ExecutionContext
55from .reflection import ObjectKind
56from .reflection import ObjectScope
57from .. import event
58from .. import exc
59from .. import pool
60from .. import util
61from ..sql import compiler
62from ..sql import dml
63from ..sql import expression
64from ..sql import type_api
65from ..sql import util as sql_util
66from ..sql._typing import is_tuple_type
67from ..sql.base import _NoArg
68from ..sql.compiler import AggregateOrderByStyle
69from ..sql.compiler import DDLCompiler
70from ..sql.compiler import InsertmanyvaluesSentinelOpts
71from ..sql.compiler import SQLCompiler
72from ..sql.elements import quoted_name
73from ..util.typing import TupleAny
74from ..util.typing import Unpack
76if typing.TYPE_CHECKING:
77 from .base import Engine
78 from .cursor import ResultFetchStrategy
79 from .interfaces import _CoreMultiExecuteParams
80 from .interfaces import _CoreSingleExecuteParams
81 from .interfaces import _DBAPICursorDescription
82 from .interfaces import _DBAPIMultiExecuteParams
83 from .interfaces import _DBAPISingleExecuteParams
84 from .interfaces import _ExecuteOptions
85 from .interfaces import _MutableCoreSingleExecuteParams
86 from .interfaces import _ParamStyle
87 from .interfaces import ConnectArgsType
88 from .interfaces import DBAPIConnection
89 from .interfaces import DBAPIModule
90 from .interfaces import DBAPIType
91 from .interfaces import IsolationLevel
92 from .interfaces import TableKey
93 from .row import Row
94 from .url import URL
95 from ..event import _ListenerFnType
96 from ..pool import Pool
97 from ..pool import PoolProxiedConnection
98 from ..sql import Executable
99 from ..sql.compiler import Compiled
100 from ..sql.compiler import Linting
101 from ..sql.compiler import ResultColumnsEntry
102 from ..sql.dml import DMLState
103 from ..sql.dml import UpdateBase
104 from ..sql.elements import BindParameter
105 from ..sql.schema import Column
106 from ..sql.sqltypes import _JSON_VALUE
107 from ..sql.type_api import _BindProcessorType
108 from ..sql.type_api import _ResultProcessorType
109 from ..sql.type_api import TypeEngine
112# When we're handed literal SQL, ensure it's a SELECT query
113SERVER_SIDE_CURSOR_RE = re.compile(r"\s*SELECT", re.I | re.UNICODE)
116(
117 CACHE_HIT,
118 CACHE_MISS,
119 CACHING_DISABLED,
120 NO_CACHE_KEY,
121 NO_DIALECT_SUPPORT,
122) = list(CacheStats)
125class _BackendsMultiReflection(Dialect):
126 """Mixin providing single-table reflection wrappers that delegate to
127 the corresponding ``get_multi_*`` methods.
129 Used by dialects that implement native multi-table reflection
130 (PostgreSQL, Oracle, MSSQL).
131 """
133 @reflection.cache
134 def has_table(
135 self,
136 connection: Connection,
137 table_name: str,
138 schema: Optional[str] = None,
139 **kw: Any,
140 ) -> bool:
141 # NOTE: assume it's a subclass of DefaultDialect
142 self._ensure_has_table_connection(connection) # type: ignore[attr-defined] # noqa: E501
143 multi_res = self.has_multi_table(
144 connection,
145 table_names=[table_name],
146 schema=schema,
147 **kw,
148 )
149 # has_multi_table returns all the input table names so it's not
150 # possible for the key to be missing
151 return dict(multi_res)[(schema, table_name)]
153 def _value_or_raise(self, data, table, schema):
154 try:
155 return dict(data)[(schema, table)]
156 except KeyError:
157 raise exc.NoSuchTableError(
158 f"{schema}.{table}" if schema else table
159 ) from None
161 @reflection.cache
162 def get_columns(self, connection, table_name, schema=None, **kw):
163 data = self.get_multi_columns(
164 connection,
165 schema=schema,
166 filter_names=[table_name],
167 scope=ObjectScope.ANY,
168 kind=ObjectKind.ANY,
169 **kw,
170 )
171 return self._value_or_raise(data, table_name, schema)
173 @reflection.cache
174 def get_table_options(self, connection, table_name, schema=None, **kw):
175 data = self.get_multi_table_options(
176 connection,
177 schema=schema,
178 filter_names=[table_name],
179 scope=ObjectScope.ANY,
180 kind=ObjectKind.ANY,
181 **kw,
182 )
183 return self._value_or_raise(data, table_name, schema)
185 @reflection.cache
186 def get_pk_constraint(self, connection, table_name, schema=None, **kw):
187 data = self.get_multi_pk_constraint(
188 connection,
189 schema=schema,
190 filter_names=[table_name],
191 scope=ObjectScope.ANY,
192 kind=ObjectKind.ANY,
193 **kw,
194 )
195 return self._value_or_raise(data, table_name, schema)
197 @reflection.cache
198 def get_foreign_keys(self, connection, table_name, schema=None, **kw):
199 data = self.get_multi_foreign_keys(
200 connection,
201 schema=schema,
202 filter_names=[table_name],
203 scope=ObjectScope.ANY,
204 kind=ObjectKind.ANY,
205 **kw,
206 )
207 return self._value_or_raise(data, table_name, schema)
209 @reflection.cache
210 def get_indexes(self, connection, table_name, schema=None, **kw):
211 data = self.get_multi_indexes(
212 connection,
213 schema=schema,
214 filter_names=[table_name],
215 scope=ObjectScope.ANY,
216 kind=ObjectKind.ANY,
217 **kw,
218 )
219 return self._value_or_raise(data, table_name, schema)
221 @reflection.cache
222 def get_unique_constraints(
223 self, connection, table_name, schema=None, **kw
224 ):
225 data = self.get_multi_unique_constraints(
226 connection,
227 schema=schema,
228 filter_names=[table_name],
229 scope=ObjectScope.ANY,
230 kind=ObjectKind.ANY,
231 **kw,
232 )
233 return self._value_or_raise(data, table_name, schema)
235 @reflection.cache
236 def get_check_constraints(self, connection, table_name, schema=None, **kw):
237 data = self.get_multi_check_constraints(
238 connection,
239 schema=schema,
240 filter_names=[table_name],
241 scope=ObjectScope.ANY,
242 kind=ObjectKind.ANY,
243 **kw,
244 )
245 return self._value_or_raise(data, table_name, schema)
247 @reflection.cache
248 def get_table_comment(self, connection, table_name, schema=None, **kw):
249 data = self.get_multi_table_comment(
250 connection,
251 schema=schema,
252 filter_names=[table_name],
253 scope=ObjectScope.ANY,
254 kind=ObjectKind.ANY,
255 **kw,
256 )
257 return self._value_or_raise(data, table_name, schema)
260class DefaultDialect(Dialect):
261 """Default implementation of Dialect"""
263 statement_compiler = compiler.SQLCompiler
264 ddl_compiler = compiler.DDLCompiler
265 type_compiler_cls = compiler.GenericTypeCompiler
267 preparer = compiler.IdentifierPreparer
268 supports_alter = True
269 supports_comments = False
270 supports_constraint_comments = False
271 inline_comments = False
272 supports_statement_cache = True
274 div_is_floordiv = True
276 bind_typing = interfaces.BindTyping.NONE
278 include_set_input_sizes: Optional[Set[Any]] = None
279 exclude_set_input_sizes: Optional[Set[Any]] = None
281 # the first value we'd get for an autoincrement column.
282 default_sequence_base = 1
284 # most DBAPIs happy with this for execute().
285 # not cx_oracle.
286 execute_sequence_format = tuple
288 supports_schemas = True
289 supports_views = True
290 supports_sequences = False
291 sequences_optional = False
292 preexecute_autoincrement_sequences = False
293 supports_identity_columns = False
294 postfetch_lastrowid = True
295 favor_returning_over_lastrowid = False
296 insert_null_pk_still_autoincrements = False
297 update_returning = False
298 delete_returning = False
299 update_returning_multifrom = False
300 delete_returning_multifrom = False
301 insert_returning = False
303 aggregate_order_by_style = AggregateOrderByStyle.INLINE
305 cte_follows_insert = False
307 supports_native_enum = False
308 supports_native_boolean = False
309 supports_native_uuid = False
310 returns_native_bytes = False
312 supports_native_json_serialization = False
313 supports_native_json_deserialization = False
314 dialect_injects_custom_json_deserializer = False
315 _json_serializer: Callable[[_JSON_VALUE], str] | None = None
317 _json_deserializer: Callable[[str], _JSON_VALUE] | None = None
319 non_native_boolean_check_constraint = True
321 supports_simple_order_by_label = True
323 tuple_in_values = False
325 connection_characteristics = util.immutabledict(
326 {
327 "isolation_level": characteristics.IsolationLevelCharacteristic(),
328 "logging_token": characteristics.LoggingTokenCharacteristic(),
329 }
330 )
332 engine_config_types: Mapping[str, Any] = util.immutabledict(
333 {
334 "pool_timeout": util.asint,
335 "echo": util.bool_or_str("debug"),
336 "echo_pool": util.bool_or_str("debug"),
337 "pool_recycle": util.asint,
338 "pool_size": util.asint,
339 "max_overflow": util.asint,
340 "future": util.asbool,
341 }
342 )
344 # if the NUMERIC type
345 # returns decimal.Decimal.
346 # *not* the FLOAT type however.
347 supports_native_decimal = False
349 name = "default"
351 # length at which to truncate
352 # any identifier.
353 max_identifier_length = 9999
354 _user_defined_max_identifier_length: Optional[int] = None
356 isolation_level: Optional[str] = None
358 # sub-categories of max_identifier_length.
359 # currently these accommodate for MySQL which allows alias names
360 # of 255 but DDL names only of 64.
361 max_index_name_length: Optional[int] = None
362 max_constraint_name_length: Optional[int] = None
364 supports_sane_rowcount = True
365 supports_sane_multi_rowcount = True
366 colspecs: MutableMapping[Type[TypeEngine[Any]], Type[TypeEngine[Any]]] = {}
367 default_paramstyle = "named"
369 supports_default_values = False
370 """dialect supports INSERT... DEFAULT VALUES syntax"""
372 supports_default_metavalue = False
373 """dialect supports INSERT... VALUES (DEFAULT) syntax"""
375 default_metavalue_token = "DEFAULT"
376 """for INSERT... VALUES (DEFAULT) syntax, the token to put in the
377 parenthesis."""
379 # not sure if this is a real thing but the compiler will deliver it
380 # if this is the only flag enabled.
381 supports_empty_insert = True
382 """dialect supports INSERT () VALUES ()"""
384 supports_multivalues_insert = False
386 use_insertmanyvalues: bool = False
388 use_insertmanyvalues_wo_returning: bool = False
390 insertmanyvalues_implicit_sentinel: InsertmanyvaluesSentinelOpts = (
391 InsertmanyvaluesSentinelOpts.NOT_SUPPORTED
392 )
394 insertmanyvalues_page_size: int = 1000
395 insertmanyvalues_max_parameters = 32700
397 supports_is_distinct_from = True
399 supports_server_side_cursors = False
401 server_side_cursors = False
403 # extra record-level locking features (#4860)
404 supports_for_update_of = False
406 server_version_info = None
408 default_schema_name: Optional[str] = None
410 # indicates symbol names are
411 # UPPERCASED if they are case insensitive
412 # within the database.
413 # if this is True, the methods normalize_name()
414 # and denormalize_name() must be provided.
415 requires_name_normalize = False
417 is_async = False
419 has_terminate = False
421 # TODO: this is not to be part of 2.0. implement rudimentary binary
422 # literals for SQLite, PostgreSQL, MySQL only within
423 # _Binary.literal_processor
424 _legacy_binary_type_literal_encoding = "utf-8"
426 @util.deprecated_params(
427 empty_in_strategy=(
428 "1.4",
429 "The :paramref:`_sa.create_engine.empty_in_strategy` keyword is "
430 "deprecated, and no longer has any effect. All IN expressions "
431 "are now rendered using "
432 'the "expanding parameter" strategy which renders a set of bound'
433 'expressions, or an "empty set" SELECT, at statement execution'
434 "time.",
435 ),
436 server_side_cursors=(
437 "1.4",
438 "The :paramref:`_sa.create_engine.server_side_cursors` parameter "
439 "is deprecated and will be removed in a future release. Please "
440 "use the "
441 ":paramref:`_engine.Connection.execution_options.stream_results` "
442 "parameter.",
443 ),
444 )
445 def __init__(
446 self,
447 paramstyle: Optional[_ParamStyle] = None,
448 isolation_level: Optional[IsolationLevel] = None,
449 dbapi: Optional[DBAPIModule] = None,
450 implicit_returning: Literal[True] = True,
451 supports_native_boolean: Optional[bool] = None,
452 max_identifier_length: Optional[int] = None,
453 label_length: Optional[int] = None,
454 insertmanyvalues_page_size: Union[_NoArg, int] = _NoArg.NO_ARG,
455 use_insertmanyvalues: Optional[bool] = None,
456 # util.deprecated_params decorator cannot render the
457 # Linting.NO_LINTING constant
458 compiler_linting: Linting = int(compiler.NO_LINTING), # type: ignore[assignment] # noqa: E501
459 server_side_cursors: bool = False,
460 skip_autocommit_rollback: bool = False,
461 **kwargs: Any,
462 ):
463 if server_side_cursors:
464 if not self.supports_server_side_cursors:
465 raise exc.ArgumentError(
466 "Dialect %s does not support server side cursors" % self
467 )
468 else:
469 self.server_side_cursors = True
471 if getattr(self, "use_setinputsizes", False):
472 util.warn_deprecated(
473 "The dialect-level use_setinputsizes attribute is "
474 "deprecated. Please use "
475 "bind_typing = BindTyping.SETINPUTSIZES",
476 "2.0",
477 )
478 self.bind_typing = interfaces.BindTyping.SETINPUTSIZES
480 self.positional = False
481 self._ischema = None
483 self.dbapi = dbapi
485 self.skip_autocommit_rollback = skip_autocommit_rollback
487 if paramstyle is not None:
488 self.paramstyle = paramstyle
489 elif self.dbapi is not None:
490 self.paramstyle = self.dbapi.paramstyle
491 else:
492 self.paramstyle = self.default_paramstyle
493 self.positional = self.paramstyle in (
494 "qmark",
495 "format",
496 "numeric",
497 "numeric_dollar",
498 )
499 self.identifier_preparer = self.preparer(self)
500 self._on_connect_isolation_level = isolation_level
502 legacy_tt_callable = getattr(self, "type_compiler", None)
503 if legacy_tt_callable is not None:
504 tt_callable = cast(
505 Type[compiler.GenericTypeCompiler],
506 self.type_compiler,
507 )
508 else:
509 tt_callable = self.type_compiler_cls
511 self.type_compiler_instance = self.type_compiler = tt_callable(self)
513 if supports_native_boolean is not None:
514 self.supports_native_boolean = supports_native_boolean
516 self._user_defined_max_identifier_length = max_identifier_length
517 if self._user_defined_max_identifier_length:
518 self.max_identifier_length = (
519 self._user_defined_max_identifier_length
520 )
521 self.label_length = label_length
522 self.compiler_linting = compiler_linting
524 if use_insertmanyvalues is not None:
525 self.use_insertmanyvalues = use_insertmanyvalues
527 if insertmanyvalues_page_size is not _NoArg.NO_ARG:
528 self.insertmanyvalues_page_size = insertmanyvalues_page_size
530 self._check_minimum_dbapi_version()
532 @property
533 @util.deprecated(
534 "2.0",
535 "full_returning is deprecated, please use insert_returning, "
536 "update_returning, delete_returning",
537 )
538 def full_returning(self):
539 return (
540 self.insert_returning
541 and self.update_returning
542 and self.delete_returning
543 )
545 @util.memoized_property
546 def insert_executemany_returning(self):
547 """Default implementation for insert_executemany_returning, if not
548 otherwise overridden by the specific dialect.
550 The default dialect determines "insert_executemany_returning" is
551 available if the dialect in use has opted into using the
552 "use_insertmanyvalues" feature. If they haven't opted into that, then
553 this attribute is False, unless the dialect in question overrides this
554 and provides some other implementation (such as the Oracle Database
555 dialects).
557 """
558 return self.insert_returning and self.use_insertmanyvalues
560 @util.memoized_property
561 def insert_executemany_returning_sort_by_parameter_order(self):
562 """Default implementation for
563 insert_executemany_returning_deterministic_order, if not otherwise
564 overridden by the specific dialect.
566 The default dialect determines "insert_executemany_returning" can have
567 deterministic order only if the dialect in use has opted into using the
568 "use_insertmanyvalues" feature, which implements deterministic ordering
569 using client side sentinel columns only by default. The
570 "insertmanyvalues" feature also features alternate forms that can
571 use server-generated PK values as "sentinels", but those are only
572 used if the :attr:`.Dialect.insertmanyvalues_implicit_sentinel`
573 bitflag enables those alternate SQL forms, which are disabled
574 by default.
576 If the dialect in use hasn't opted into that, then this attribute is
577 False, unless the dialect in question overrides this and provides some
578 other implementation (such as the Oracle Database dialects).
580 """
581 return self.insert_returning and self.use_insertmanyvalues
583 update_executemany_returning = False
584 delete_executemany_returning = False
586 @util.memoized_property
587 def loaded_dbapi(self) -> DBAPIModule:
588 if self.dbapi is None:
589 raise exc.NoDBAPILoaded(
590 f"Dialect {self} does not have a Python DBAPI established "
591 "and cannot be used for actual database interaction"
592 )
593 return self.dbapi
595 @util.memoized_property
596 def dbapi_version(self) -> util.VersionInfo:
597 # memoization applies to a successfully determined version only;
598 # memoized_property does not cache when the function raises, so a
599 # DBAPI which is established after this dialect was constructed is
600 # still picked up
601 if self.dbapi is None:
602 raise exc.NoDBAPILoaded(
603 f"Dialect {self.name}+{self.driver} has no DBAPI module "
604 "loaded; no DBAPI version is available"
605 )
607 try:
608 version = self.retrieve_dbapi_version(self.dbapi)
609 except NotImplementedError as ne:
610 raise NotImplementedError(
611 f"Dialect {self.name}+{self.driver} does not implement "
612 "retrieve_dbapi_version(); no DBAPI version is available"
613 ) from ne
615 if not version:
616 # the DBAPI is loaded but publishes no version of its own; as
617 # with a DBAPI that isn't loaded, this is not an error on the
618 # part of the dialect
619 # asyncio dialects have a wrapper object here rather than a
620 # module, which has no __name__
621 dbapi_name = getattr(self.dbapi, "__name__", self.driver)
622 raise exc.NoDBAPILoaded(
623 f"Dialect {self.name}+{self.driver} could not determine a "
624 f"version for its DBAPI module {dbapi_name!r}"
625 )
627 return version
629 @property
630 def _dbapi_version_or_none(self) -> Optional[util.VersionInfo]:
631 """:attr:`.Dialect.dbapi_version`, or None if it can't be
632 determined.
634 For use by dialect startup checks, which run before a DBAPI is
635 necessarily present and which must not fail when no version is
636 available. Only :class:`.exc.NoDBAPILoaded` is accommodated;
637 ``NotImplementedError``, indicating a dialect which does not
638 implement :meth:`.Dialect.retrieve_dbapi_version` at all, is a bug
639 in that dialect and is allowed to propagate. Deliberately not
640 memoized itself; the underlying :attr:`.Dialect.dbapi_version`
641 memoizes the success case.
643 """
644 try:
645 return self.dbapi_version
646 except exc.NoDBAPILoaded:
647 return None
649 def _check_minimum_dbapi_version(self) -> None:
650 """Enforce :attr:`.Dialect.minimum_dbapi_version`, if present.
652 Takes place as the dialect is constructed. No check occurs when
653 the version of the DBAPI is not available at all.
655 """
656 minimum = self.minimum_dbapi_version
657 if minimum is None:
658 return
660 version = self._dbapi_version_or_none
661 if version is not None and version < minimum:
662 dbapi_name = getattr(self.dbapi, "__name__", self.driver)
663 raise exc.InvalidRequestError(
664 f"Dialect {self.name}+{self.driver} requires version "
665 f"{minimum} or greater of the {dbapi_name} DBAPI; "
666 f"version {version} is installed"
667 )
669 @util.memoized_property
670 def _bind_typing_render_casts(self):
671 return self.bind_typing is interfaces.BindTyping.RENDER_CASTS
673 def _ensure_has_table_connection(self, arg: Connection) -> None:
674 if not isinstance(arg, Connection):
675 raise exc.ArgumentError(
676 "The argument passed to Dialect.has_table() should be a "
677 "%s, got %s. "
678 "Additionally, the Dialect.has_table() method is for "
679 "internal dialect "
680 "use only; please use "
681 "``inspect(some_engine).has_table(<tablename>>)`` "
682 "for public API use." % (Connection, type(arg))
683 )
685 @util.memoized_property
686 def _supports_statement_cache(self):
687 ssc = self.__class__.__dict__.get("supports_statement_cache", None)
688 if ssc is None:
689 util.warn(
690 "Dialect %s:%s will not make use of SQL compilation caching "
691 "as it does not set the 'supports_statement_cache' attribute "
692 "to ``True``. This can have "
693 "significant performance implications including some "
694 "performance degradations in comparison to prior SQLAlchemy "
695 "versions. Dialect maintainers should seek to set this "
696 "attribute to True after appropriate development and testing "
697 "for SQLAlchemy 1.4 caching support. Alternatively, this "
698 "attribute may be set to False which will disable this "
699 "warning." % (self.name, self.driver),
700 code="cprf",
701 )
703 return bool(ssc)
705 @util.memoized_property
706 def _type_memos(self):
707 return weakref.WeakKeyDictionary()
709 @property
710 def dialect_description(self): # type: ignore[override]
711 return self.name + "+" + self.driver
713 @property
714 def supports_sane_rowcount_returning(self):
715 """True if this dialect supports sane rowcount even if RETURNING is
716 in use.
718 For dialects that don't support RETURNING, this is synonymous with
719 ``supports_sane_rowcount``.
721 """
722 return self.supports_sane_rowcount
724 @classmethod
725 def get_pool_class(cls, url: URL) -> Type[Pool]:
726 default: Type[pool.Pool]
727 if cls.is_async:
728 default = pool.AsyncAdaptedQueuePool
729 else:
730 default = pool.QueuePool
732 return getattr(cls, "poolclass", default)
734 def get_dialect_pool_class(self, url: URL) -> Type[Pool]:
735 return self.get_pool_class(url)
737 @classmethod
738 def load_provisioning(cls):
739 package = ".".join(cls.__module__.split(".")[0:-1])
740 try:
741 __import__(package + ".provision")
742 except ImportError:
743 pass
745 def _builtin_onconnect(self) -> Optional[_ListenerFnType]:
746 if self._on_connect_isolation_level is not None:
748 def builtin_connect(dbapi_conn, conn_rec):
749 self._assert_and_set_isolation_level(
750 dbapi_conn, self._on_connect_isolation_level
751 )
753 return builtin_connect
754 else:
755 return None
757 def initialize(self, connection: Connection) -> None:
758 try:
759 self.server_version_info = self._get_server_version_info(
760 connection
761 )
762 except NotImplementedError:
763 self.server_version_info = None
764 try:
765 self.default_schema_name = self._get_default_schema_name(
766 connection
767 )
768 except NotImplementedError:
769 self.default_schema_name = None
771 try:
772 self.default_isolation_level = self.get_default_isolation_level(
773 connection.connection.dbapi_connection
774 )
775 except NotImplementedError:
776 self.default_isolation_level = None
778 if not self._user_defined_max_identifier_length:
779 max_ident_length = self._check_max_identifier_length(connection)
780 if max_ident_length:
781 self.max_identifier_length = max_ident_length
783 if (
784 self.label_length
785 and self.label_length > self.max_identifier_length
786 ):
787 raise exc.ArgumentError(
788 "Label length of %d is greater than this dialect's"
789 " maximum identifier length of %d"
790 % (self.label_length, self.max_identifier_length)
791 )
793 def on_connect(self) -> Optional[Callable[[Any], None]]:
794 # inherits the docstring from interfaces.Dialect.on_connect
795 return None
797 def _check_max_identifier_length(self, connection):
798 """Perform a connection / server version specific check to determine
799 the max_identifier_length.
801 If the dialect's class level max_identifier_length should be used,
802 can return None.
804 """
805 return None
807 def get_default_isolation_level(self, dbapi_conn):
808 """Given a DBAPI connection, return its isolation level, or
809 a default isolation level if one cannot be retrieved.
811 May be overridden by subclasses in order to provide a
812 "fallback" isolation level for databases that cannot reliably
813 retrieve the actual isolation level.
815 By default, calls the :meth:`_engine.Interfaces.get_isolation_level`
816 method, propagating any exceptions raised.
818 """
819 return self.get_isolation_level(dbapi_conn)
821 def type_descriptor(self, typeobj):
822 """Provide a database-specific :class:`.TypeEngine` object, given
823 the generic object which comes from the types module.
825 This method looks for a dictionary called
826 ``colspecs`` as a class or instance-level variable,
827 and passes on to :func:`_types.adapt_type`.
829 """
830 return type_api.adapt_type(typeobj, self.colspecs)
832 def has_index(self, connection, table_name, index_name, schema=None, **kw):
833 if not self.has_table(connection, table_name, schema=schema, **kw):
834 return False
835 for idx in self.get_indexes(
836 connection, table_name, schema=schema, **kw
837 ):
838 if idx["name"] == index_name:
839 return True
840 else:
841 return False
843 def has_schema(
844 self, connection: Connection, schema_name: str, **kw: Any
845 ) -> bool:
846 return schema_name in self.get_schema_names(connection, **kw)
848 def validate_identifier(self, ident: str) -> None:
849 if len(ident) > self.max_identifier_length:
850 raise exc.IdentifierError(
851 "Identifier '%s' exceeds maximum length of %d characters"
852 % (ident, self.max_identifier_length)
853 )
855 def connect(self, *cargs: Any, **cparams: Any) -> DBAPIConnection:
856 # inherits the docstring from interfaces.Dialect.connect
857 return self.loaded_dbapi.connect(*cargs, **cparams) # type: ignore[no-any-return] # NOQA: E501
859 def create_connect_args(self, url: URL) -> ConnectArgsType:
860 # inherits the docstring from interfaces.Dialect.create_connect_args
861 opts = url.translate_connect_args()
862 opts.update(url.query)
863 return ([], opts)
865 def set_engine_execution_options(
866 self, engine: Engine, opts: Mapping[str, Any]
867 ) -> None:
868 supported_names = set(self.connection_characteristics).intersection(
869 opts
870 )
871 if supported_names:
872 characteristics: Mapping[str, Any] = util.immutabledict(
873 (name, opts[name]) for name in supported_names
874 )
876 @event.listens_for(engine, "engine_connect")
877 def set_connection_characteristics(connection):
878 self._set_connection_characteristics(
879 connection, characteristics
880 )
882 def set_connection_execution_options(
883 self, connection: Connection, opts: Mapping[str, Any]
884 ) -> None:
885 supported_names = set(self.connection_characteristics).intersection(
886 opts
887 )
888 if supported_names:
889 characteristics: Mapping[str, Any] = util.immutabledict(
890 (name, opts[name]) for name in supported_names
891 )
892 self._set_connection_characteristics(connection, characteristics)
894 def _set_connection_characteristics(self, connection, characteristics):
895 characteristic_values = [
896 (name, self.connection_characteristics[name], value)
897 for name, value in characteristics.items()
898 ]
900 if connection.in_transaction():
901 trans_objs = [
902 (name, obj)
903 for name, obj, _ in characteristic_values
904 if obj.transactional
905 ]
906 if trans_objs:
907 raise exc.InvalidRequestError(
908 "This connection has already initialized a SQLAlchemy "
909 "Transaction() object via begin() or autobegin; "
910 "%s may not be altered unless rollback() or commit() "
911 "is called first."
912 % (", ".join(name for name, obj in trans_objs))
913 )
915 dbapi_connection = connection.connection.dbapi_connection
916 for _, characteristic, value in characteristic_values:
917 characteristic.set_connection_characteristic(
918 self, connection, dbapi_connection, value
919 )
920 connection.connection._connection_record.finalize_callback.append(
921 functools.partial(self._reset_characteristics, characteristics)
922 )
924 def _reset_characteristics(self, characteristics, dbapi_connection):
925 for characteristic_name in characteristics:
926 characteristic = self.connection_characteristics[
927 characteristic_name
928 ]
929 characteristic.reset_characteristic(self, dbapi_connection)
931 def do_begin(self, dbapi_connection):
932 pass
934 def do_rollback(self, dbapi_connection):
935 if self.skip_autocommit_rollback and self.detect_autocommit_setting(
936 dbapi_connection
937 ):
938 return
939 dbapi_connection.rollback()
941 def do_commit(self, dbapi_connection):
942 dbapi_connection.commit()
944 def do_terminate(self, dbapi_connection):
945 self.do_close(dbapi_connection)
947 def do_close(self, dbapi_connection):
948 dbapi_connection.close()
950 @util.memoized_property
951 def _dialect_specific_select_one(self):
952 return str(expression.select(1).compile(dialect=self))
954 def _do_ping_w_event(self, dbapi_connection: DBAPIConnection) -> bool:
955 try:
956 return self.do_ping(dbapi_connection)
957 except self.loaded_dbapi.Error as err:
958 is_disconnect = self.is_disconnect(err, dbapi_connection, None)
960 if self._has_events:
961 try:
962 Connection._handle_dbapi_exception_noconnection(
963 err,
964 self,
965 is_disconnect=is_disconnect,
966 invalidate_pool_on_disconnect=False,
967 is_pre_ping=True,
968 )
969 except exc.StatementError as new_err:
970 is_disconnect = new_err.connection_invalidated
972 if is_disconnect:
973 return False
974 else:
975 raise
977 def do_ping(self, dbapi_connection: DBAPIConnection) -> bool:
978 cursor = dbapi_connection.cursor()
979 try:
980 cursor.execute(self._dialect_specific_select_one)
981 finally:
982 cursor.close()
983 return True
985 def create_xid(self):
986 """Create a random two-phase transaction ID.
988 This id will be passed to do_begin_twophase(), do_rollback_twophase(),
989 do_commit_twophase(). Its format is unspecified.
990 """
992 return "_sa_%032x" % random.randint(0, 2**128)
994 def do_savepoint(self, connection, name):
995 connection.execute(expression.SavepointClause(name))
997 def do_rollback_to_savepoint(self, connection, name):
998 connection.execute(expression.RollbackToSavepointClause(name))
1000 def do_release_savepoint(self, connection, name):
1001 connection.execute(expression.ReleaseSavepointClause(name))
1003 def _deliver_insertmanyvalues_batches(
1004 self,
1005 connection,
1006 cursor,
1007 statement,
1008 parameters,
1009 generic_setinputsizes,
1010 context,
1011 ):
1012 context = cast(DefaultExecutionContext, context)
1013 compiled = cast(SQLCompiler, context.compiled)
1015 _composite_sentinel_proc: Sequence[
1016 Optional[_ResultProcessorType[Any]]
1017 ] = ()
1018 _scalar_sentinel_proc: Optional[_ResultProcessorType[Any]] = None
1019 _sentinel_proc_initialized: bool = False
1021 compiled_parameters = context.compiled_parameters
1023 imv = compiled._insertmanyvalues
1024 assert imv is not None
1026 is_returning: Final[bool] = bool(compiled.effective_returning)
1027 batch_size = context.execution_options.get(
1028 "insertmanyvalues_page_size", self.insertmanyvalues_page_size
1029 )
1031 if compiled.schema_translate_map:
1032 schema_translate_map = context.execution_options.get(
1033 "schema_translate_map", {}
1034 )
1035 else:
1036 schema_translate_map = None
1038 if is_returning:
1039 result: Optional[List[Any]] = []
1040 context._insertmanyvalues_rows = result
1042 sort_by_parameter_order = imv.sort_by_parameter_order
1044 else:
1045 sort_by_parameter_order = False
1046 result = None
1048 for imv_batch in compiled._deliver_insertmanyvalues_batches(
1049 statement,
1050 parameters,
1051 compiled_parameters,
1052 generic_setinputsizes,
1053 batch_size,
1054 sort_by_parameter_order,
1055 schema_translate_map,
1056 ):
1057 yield imv_batch
1059 if is_returning:
1061 try:
1062 rows = context.fetchall_for_returning(cursor)
1063 except BaseException as be:
1064 connection._handle_dbapi_exception(
1065 be,
1066 sql_util._long_statement(imv_batch.replaced_statement),
1067 imv_batch.replaced_parameters,
1068 None,
1069 context,
1070 is_sub_exec=True,
1071 )
1073 # I would have thought "is_returning: Final[bool]"
1074 # would have assured this but pylance thinks not
1075 assert result is not None
1077 if imv.num_sentinel_columns and not imv_batch.is_downgraded:
1078 composite_sentinel = imv.num_sentinel_columns > 1
1079 if imv.implicit_sentinel:
1080 # for implicit sentinel, which is currently single-col
1081 # integer autoincrement, do a simple sort.
1082 assert not composite_sentinel
1083 result.extend(
1084 sorted(rows, key=operator.itemgetter(-1))
1085 )
1086 continue
1088 # otherwise, create dictionaries to match up batches
1089 # with parameters
1090 assert imv.sentinel_param_keys
1091 assert imv.sentinel_columns
1093 _nsc = imv.num_sentinel_columns
1095 if not _sentinel_proc_initialized:
1096 if composite_sentinel:
1097 _composite_sentinel_proc = [
1098 col.type._cached_result_processor(
1099 self, cursor_desc[1]
1100 )
1101 for col, cursor_desc in zip(
1102 imv.sentinel_columns,
1103 cursor.description[-_nsc:],
1104 )
1105 ]
1106 else:
1107 _scalar_sentinel_proc = (
1108 imv.sentinel_columns[0]
1109 ).type._cached_result_processor(
1110 self, cursor.description[-1][1]
1111 )
1112 _sentinel_proc_initialized = True
1114 rows_by_sentinel: Union[
1115 Dict[Tuple[Any, ...], Any],
1116 Dict[Any, Any],
1117 ]
1119 if composite_sentinel:
1120 rows_by_sentinel = {
1121 tuple(
1122 (proc(val) if proc else val)
1123 for val, proc in zip(
1124 row[-_nsc:], _composite_sentinel_proc
1125 )
1126 ): row
1127 for row in rows
1128 }
1129 elif _scalar_sentinel_proc:
1130 rows_by_sentinel = {
1131 _scalar_sentinel_proc(row[-1]): row for row in rows
1132 }
1133 else:
1134 rows_by_sentinel = {row[-1]: row for row in rows}
1136 if len(rows_by_sentinel) != len(imv_batch.batch):
1137 # see test_insert_exec.py::
1138 # IMVSentinelTest::test_sentinel_incorrect_rowcount
1139 # for coverage / demonstration
1140 raise exc.InvalidRequestError(
1141 f"Sentinel-keyed result set did not produce "
1142 f"correct number of rows {len(imv_batch.batch)}; "
1143 "produced "
1144 f"{len(rows_by_sentinel)}. Please ensure the "
1145 "sentinel column is fully unique and populated in "
1146 "all cases."
1147 )
1149 try:
1150 ordered_rows = [
1151 rows_by_sentinel[sentinel_keys]
1152 for sentinel_keys in imv_batch.sentinel_values
1153 ]
1154 except KeyError as ke:
1155 # see test_insert_exec.py::
1156 # IMVSentinelTest::test_sentinel_cant_match_keys
1157 # for coverage / demonstration
1158 raise exc.InvalidRequestError(
1159 f"Can't match sentinel values in result set to "
1160 f"parameter sets; key {ke.args[0]!r} was not "
1161 "found. "
1162 "There may be a mismatch between the datatype "
1163 "passed to the DBAPI driver vs. that which it "
1164 "returns in a result row. Ensure the given "
1165 "Python value matches the expected result type "
1166 "*exactly*, taking care to not rely upon implicit "
1167 "conversions which may occur such as when using "
1168 "strings in place of UUID or integer values, etc. "
1169 ) from ke
1171 result.extend(ordered_rows)
1173 else:
1174 result.extend(rows)
1176 def do_executemany(self, cursor, statement, parameters, context=None):
1177 cursor.executemany(statement, parameters)
1179 def do_execute(self, cursor, statement, parameters, context=None):
1180 cursor.execute(statement, parameters)
1182 def do_execute_no_params(self, cursor, statement, context=None):
1183 cursor.execute(statement)
1185 def is_disconnect(
1186 self,
1187 e: DBAPIModule.Error,
1188 connection: Union[
1189 pool.PoolProxiedConnection, interfaces.DBAPIConnection, None
1190 ],
1191 cursor: Optional[interfaces.DBAPICursor],
1192 ) -> bool:
1193 return False
1195 @util.memoized_instancemethod
1196 def _gen_allowed_isolation_levels(self, dbapi_conn):
1197 try:
1198 raw_levels = list(self.get_isolation_level_values(dbapi_conn))
1199 except NotImplementedError:
1200 return None
1201 else:
1202 normalized_levels = [
1203 level.replace("_", " ").upper() for level in raw_levels
1204 ]
1205 if raw_levels != normalized_levels:
1206 raise ValueError(
1207 f"Dialect {self.name!r} get_isolation_level_values() "
1208 f"method should return names as UPPERCASE using spaces, "
1209 f"not underscores; got "
1210 f"{sorted(set(raw_levels).difference(normalized_levels))}"
1211 )
1212 return tuple(normalized_levels)
1214 def _assert_and_set_isolation_level(self, dbapi_conn, level):
1215 level = level.replace("_", " ").upper()
1217 _allowed_isolation_levels = self._gen_allowed_isolation_levels(
1218 dbapi_conn
1219 )
1220 if (
1221 _allowed_isolation_levels
1222 and level not in _allowed_isolation_levels
1223 ):
1224 raise exc.ArgumentError(
1225 f"Invalid value {level!r} for isolation_level. "
1226 f"Valid isolation levels for {self.name!r} are "
1227 f"{', '.join(_allowed_isolation_levels)}"
1228 )
1230 self.set_isolation_level(dbapi_conn, level)
1232 def reset_isolation_level(self, dbapi_conn):
1233 if self._on_connect_isolation_level is not None:
1234 assert (
1235 self._on_connect_isolation_level == "AUTOCOMMIT"
1236 or self._on_connect_isolation_level
1237 == self.default_isolation_level
1238 )
1239 self._assert_and_set_isolation_level(
1240 dbapi_conn, self._on_connect_isolation_level
1241 )
1242 else:
1243 assert self.default_isolation_level is not None
1244 self._assert_and_set_isolation_level(
1245 dbapi_conn,
1246 self.default_isolation_level,
1247 )
1249 def normalize_name(self, name):
1250 if name is None:
1251 return None
1253 name_lower = name.lower()
1254 name_upper = name.upper()
1256 if name_upper == name_lower:
1257 # name has no upper/lower conversion, e.g. non-european characters.
1258 # return unchanged
1259 return name
1260 elif name_upper == name and not (
1261 self.identifier_preparer._requires_quotes
1262 )(name_lower):
1263 # name is all uppercase and doesn't require quoting; normalize
1264 # to all lower case
1265 return name_lower
1266 elif name_lower == name:
1267 # name is all lower case, which if denormalized means we need to
1268 # force quoting on it
1269 return quoted_name(name, quote=True)
1270 else:
1271 # name is mixed case, means it will be quoted in SQL when used
1272 # later, no normalizes
1273 return name
1275 def denormalize_name(self, name):
1276 if name is None:
1277 return None
1279 name_lower = name.lower()
1280 name_upper = name.upper()
1282 if name_upper == name_lower:
1283 # name has no upper/lower conversion, e.g. non-european characters.
1284 # return unchanged
1285 return name
1286 elif name_lower == name and not (
1287 self.identifier_preparer._requires_quotes
1288 )(name_lower):
1289 name = name_upper
1290 return name
1292 def get_driver_connection(self, connection: DBAPIConnection) -> Any:
1293 return connection
1295 def _overrides_default(self, method):
1296 return (
1297 getattr(type(self), method).__code__
1298 is not getattr(DefaultDialect, method).__code__
1299 )
1301 def _default_multi_reflect(
1302 self,
1303 single_tbl_method,
1304 connection,
1305 kind,
1306 schema,
1307 filter_names,
1308 scope,
1309 **kw,
1310 ):
1311 names_fns = []
1312 temp_names_fns = []
1313 if ObjectKind.TABLE in kind:
1314 names_fns.append(self.get_table_names)
1315 temp_names_fns.append(self.get_temp_table_names)
1316 if ObjectKind.VIEW in kind:
1317 names_fns.append(self.get_view_names)
1318 temp_names_fns.append(self.get_temp_view_names)
1319 if ObjectKind.MATERIALIZED_VIEW in kind:
1320 names_fns.append(self.get_materialized_view_names)
1321 # no temp materialized view at the moment
1322 # temp_names_fns.append(self.get_temp_materialized_view_names)
1324 unreflectable = kw.pop("unreflectable", {})
1326 if (
1327 filter_names
1328 and scope is ObjectScope.ANY
1329 and kind is ObjectKind.ANY
1330 ):
1331 # if names are given and no qualification on type of table
1332 # (i.e. the Table(..., autoload) case), take the names as given,
1333 # don't run names queries. If a table does not exit
1334 # NoSuchTableError is raised and it's skipped
1336 # this also suits the case for mssql where we can reflect
1337 # individual temp tables but there's no temp_names_fn
1338 names = filter_names
1339 else:
1340 names = []
1341 name_kw = {"schema": schema, **kw}
1342 fns = []
1343 if ObjectScope.DEFAULT in scope:
1344 fns.extend(names_fns)
1345 if ObjectScope.TEMPORARY in scope:
1346 fns.extend(temp_names_fns)
1348 for fn in fns:
1349 try:
1350 names.extend(fn(connection, **name_kw))
1351 except NotImplementedError:
1352 pass
1354 if filter_names:
1355 filter_names = set(filter_names)
1357 # iterate over all the tables/views and call the single table method
1358 for table in names:
1359 if not filter_names or table in filter_names:
1360 key = (schema, table)
1361 try:
1362 yield (
1363 key,
1364 single_tbl_method(
1365 connection, table, schema=schema, **kw
1366 ),
1367 )
1368 except exc.UnreflectableTableError as err:
1369 if key not in unreflectable:
1370 unreflectable[key] = err
1371 except exc.NoSuchTableError:
1372 pass
1374 def has_multi_table(
1375 self,
1376 connection: Connection,
1377 table_names: Sequence[str],
1378 schema: Optional[str] = None,
1379 **kw: Any,
1380 ) -> Iterable[tuple[TableKey, bool]]:
1381 for table_name in table_names:
1382 exist = self.has_table(connection, table_name, schema=schema, **kw)
1383 yield (schema, table_name), exist
1385 def get_multi_table_options(self, connection, **kw):
1386 return self._default_multi_reflect(
1387 self.get_table_options, connection, **kw
1388 )
1390 def get_multi_columns(self, connection, **kw):
1391 return self._default_multi_reflect(self.get_columns, connection, **kw)
1393 def get_multi_pk_constraint(self, connection, **kw):
1394 return self._default_multi_reflect(
1395 self.get_pk_constraint, connection, **kw
1396 )
1398 def get_multi_foreign_keys(self, connection, **kw):
1399 return self._default_multi_reflect(
1400 self.get_foreign_keys, connection, **kw
1401 )
1403 def get_multi_indexes(self, connection, **kw):
1404 return self._default_multi_reflect(self.get_indexes, connection, **kw)
1406 def get_multi_unique_constraints(self, connection, **kw):
1407 return self._default_multi_reflect(
1408 self.get_unique_constraints, connection, **kw
1409 )
1411 def get_multi_check_constraints(self, connection, **kw):
1412 return self._default_multi_reflect(
1413 self.get_check_constraints, connection, **kw
1414 )
1416 def get_multi_table_comment(self, connection, **kw):
1417 return self._default_multi_reflect(
1418 self.get_table_comment, connection, **kw
1419 )
1422class StrCompileDialect(DefaultDialect):
1423 statement_compiler = compiler.StrSQLCompiler
1424 ddl_compiler = compiler.DDLCompiler
1425 type_compiler_cls = compiler.StrSQLTypeCompiler
1426 preparer = compiler.IdentifierPreparer
1428 insert_returning = True
1429 update_returning = True
1430 delete_returning = True
1432 supports_statement_cache = True
1434 supports_identity_columns = True
1436 supports_sequences = True
1437 sequences_optional = True
1438 preexecute_autoincrement_sequences = False
1440 supports_native_boolean = True
1442 supports_multivalues_insert = True
1443 supports_simple_order_by_label = True
1446class DefaultExecutionContext(ExecutionContext):
1447 isinsert = False
1448 isupdate = False
1449 isdelete = False
1450 is_crud = False
1451 is_text = False
1452 isddl = False
1454 execute_style: ExecuteStyle = ExecuteStyle.EXECUTE
1456 compiled: Optional[Compiled] = None
1457 result_column_struct: Optional[
1458 Tuple[List[ResultColumnsEntry], bool, bool, bool, bool]
1459 ] = None
1460 returned_default_rows: Optional[Sequence[Row[Unpack[TupleAny]]]] = None
1462 execution_options: _ExecuteOptions = util.EMPTY_DICT
1464 cursor_fetch_strategy = _cursor._DEFAULT_FETCH
1466 invoked_statement: Optional[Executable] = None
1468 _is_implicit_returning = False
1469 _is_explicit_returning = False
1470 _is_supplemental_returning = False
1471 _is_server_side = False
1473 _soft_closed = False
1475 _rowcount: Optional[int] = None
1477 # a hook for SQLite's translation of
1478 # result column names
1479 # NOTE: pyhive is using this hook, can't remove it :(
1480 _translate_colname: Optional[
1481 Callable[[str], Tuple[str, Optional[str]]]
1482 ] = None
1484 _expanded_parameters: Mapping[str, List[str]] = util.immutabledict()
1485 """used by set_input_sizes().
1487 This collection comes from ``ExpandedState.parameter_expansion``.
1489 """
1491 cache_hit = NO_CACHE_KEY
1493 root_connection: Connection
1494 _dbapi_connection: PoolProxiedConnection
1495 dialect: Dialect
1496 unicode_statement: str
1497 cursor: DBAPICursor
1498 compiled_parameters: List[_MutableCoreSingleExecuteParams]
1499 parameters: _DBAPIMultiExecuteParams
1500 extracted_parameters: Optional[Sequence[BindParameter[Any]]]
1502 _empty_dict_params = cast("Mapping[str, Any]", util.EMPTY_DICT)
1504 _insertmanyvalues_rows: Optional[List[Tuple[Any, ...]]] = None
1505 _num_sentinel_cols: int = 0
1507 @classmethod
1508 def _init_ddl(
1509 cls,
1510 dialect: Dialect,
1511 connection: Connection,
1512 dbapi_connection: PoolProxiedConnection,
1513 execution_options: _ExecuteOptions,
1514 compiled_ddl: DDLCompiler,
1515 ) -> ExecutionContext:
1516 """Initialize execution context for an ExecutableDDLElement
1517 construct."""
1519 self = cls.__new__(cls)
1520 self.root_connection = connection
1521 self._dbapi_connection = dbapi_connection
1522 self.dialect = connection.dialect
1524 self.compiled = compiled = compiled_ddl
1525 self.isddl = True
1527 self.execution_options = execution_options
1529 self.unicode_statement = str(compiled)
1530 if compiled.schema_translate_map:
1531 schema_translate_map = self.execution_options.get(
1532 "schema_translate_map", {}
1533 )
1535 rst = compiled.preparer._render_schema_translates
1536 self.unicode_statement = rst(
1537 self.unicode_statement, schema_translate_map
1538 )
1540 self.statement = self.unicode_statement
1542 self.cursor = self.create_cursor()
1543 self.compiled_parameters = []
1545 if dialect.positional:
1546 self.parameters = [dialect.execute_sequence_format()]
1547 else:
1548 self.parameters = [self._empty_dict_params]
1550 return self
1552 @classmethod
1553 def _init_compiled(
1554 cls,
1555 dialect: Dialect,
1556 connection: Connection,
1557 dbapi_connection: PoolProxiedConnection,
1558 execution_options: _ExecuteOptions,
1559 compiled: SQLCompiler,
1560 parameters: _CoreMultiExecuteParams,
1561 invoked_statement: Executable,
1562 extracted_parameters: Optional[Sequence[BindParameter[Any]]],
1563 cache_hit: CacheStats = CacheStats.CACHING_DISABLED,
1564 param_dict: _CoreSingleExecuteParams | None = None,
1565 ) -> ExecutionContext:
1566 """Initialize execution context for a Compiled construct."""
1568 self = cls.__new__(cls)
1569 self.root_connection = connection
1570 self._dbapi_connection = dbapi_connection
1571 self.dialect = connection.dialect
1572 self.extracted_parameters = extracted_parameters
1573 self.invoked_statement = invoked_statement
1574 self.compiled = compiled
1575 self.cache_hit = cache_hit
1577 self.execution_options = execution_options
1579 self.result_column_struct = (
1580 compiled._result_columns,
1581 compiled._ordered_columns,
1582 compiled._textual_ordered_columns,
1583 compiled._ad_hoc_textual,
1584 compiled._loose_column_name_matching,
1585 )
1587 self.isinsert = ii = compiled.isinsert
1588 self.isupdate = iu = compiled.isupdate
1589 self.isdelete = id_ = compiled.isdelete
1590 self.is_text = compiled.isplaintext
1592 if ii or iu or id_:
1593 dml_statement = compiled.compile_state.statement # type: ignore[union-attr] # noqa: E501
1594 if TYPE_CHECKING:
1595 assert isinstance(dml_statement, UpdateBase)
1596 self.is_crud = True
1597 self._is_explicit_returning = ier = bool(dml_statement._returning)
1598 self._is_implicit_returning = iir = bool(
1599 compiled.implicit_returning
1600 )
1601 if iir and dml_statement._supplemental_returning:
1602 self._is_supplemental_returning = True
1604 # dont mix implicit and explicit returning
1605 assert not (iir and ier)
1607 if (ier or iir) and compiled.for_executemany:
1608 if ii and not self.dialect.insert_executemany_returning:
1609 raise exc.InvalidRequestError(
1610 f"Dialect {self.dialect.dialect_description} with "
1611 f"current server capabilities does not support "
1612 "INSERT..RETURNING when executemany is used"
1613 )
1614 elif (
1615 ii
1616 and dml_statement._sort_by_parameter_order
1617 and not self.dialect.insert_executemany_returning_sort_by_parameter_order # noqa: E501
1618 ):
1619 raise exc.InvalidRequestError(
1620 f"Dialect {self.dialect.dialect_description} with "
1621 f"current server capabilities does not support "
1622 "INSERT..RETURNING with deterministic row ordering "
1623 "when executemany is used"
1624 )
1625 elif (
1626 ii
1627 and self.dialect.use_insertmanyvalues
1628 and not compiled._insertmanyvalues
1629 ):
1630 raise exc.InvalidRequestError(
1631 'Statement does not have "insertmanyvalues" '
1632 "enabled, can't use INSERT..RETURNING with "
1633 "executemany in this case."
1634 )
1635 elif iu and not self.dialect.update_executemany_returning:
1636 raise exc.InvalidRequestError(
1637 f"Dialect {self.dialect.dialect_description} with "
1638 f"current server capabilities does not support "
1639 "UPDATE..RETURNING when executemany is used"
1640 )
1641 elif id_ and not self.dialect.delete_executemany_returning:
1642 raise exc.InvalidRequestError(
1643 f"Dialect {self.dialect.dialect_description} with "
1644 f"current server capabilities does not support "
1645 "DELETE..RETURNING when executemany is used"
1646 )
1648 if not parameters:
1649 self.compiled_parameters = [
1650 compiled.construct_params(
1651 extracted_parameters=extracted_parameters,
1652 escape_names=False,
1653 _collected_params=param_dict,
1654 )
1655 ]
1656 else:
1657 self.compiled_parameters = [
1658 compiled.construct_params(
1659 m,
1660 escape_names=False,
1661 _group_number=grp,
1662 extracted_parameters=extracted_parameters,
1663 _collected_params=param_dict,
1664 )
1665 for grp, m in enumerate(parameters)
1666 ]
1668 if len(parameters) > 1:
1669 if self.isinsert and compiled._insertmanyvalues:
1670 self.execute_style = ExecuteStyle.INSERTMANYVALUES
1672 imv = compiled._insertmanyvalues
1673 if imv.sentinel_columns is not None:
1674 self._num_sentinel_cols = imv.num_sentinel_columns
1675 else:
1676 self.execute_style = ExecuteStyle.EXECUTEMANY
1678 self.unicode_statement = compiled.string
1680 self.cursor = self.create_cursor()
1682 if self.compiled.insert_prefetch or self.compiled.update_prefetch:
1683 self._process_execute_defaults()
1685 processors = compiled._bind_processors
1687 flattened_processors: Mapping[
1688 str, _BindProcessorType[Any]
1689 ] = processors # type: ignore[assignment]
1691 if compiled.literal_execute_params or compiled.post_compile_params:
1692 if self.executemany:
1693 raise exc.InvalidRequestError(
1694 "'literal_execute' or 'expanding' parameters can't be "
1695 "used with executemany()"
1696 )
1698 expanded_state = compiled._process_parameters_for_postcompile(
1699 self.compiled_parameters[0]
1700 )
1702 # re-assign self.unicode_statement
1703 self.unicode_statement = expanded_state.statement
1705 self._expanded_parameters = expanded_state.parameter_expansion
1707 flattened_processors = dict(processors) # type: ignore[arg-type]
1708 flattened_processors.update(expanded_state.processors)
1709 positiontup = expanded_state.positiontup
1710 elif compiled.positional:
1711 positiontup = self.compiled.positiontup
1712 else:
1713 positiontup = None
1715 if compiled.schema_translate_map:
1716 schema_translate_map = self.execution_options.get(
1717 "schema_translate_map", {}
1718 )
1719 rst = compiled.preparer._render_schema_translates
1720 self.unicode_statement = rst(
1721 self.unicode_statement, schema_translate_map
1722 )
1724 # final self.unicode_statement is now assigned, encode if needed
1725 # by dialect
1726 self.statement = self.unicode_statement
1728 # Convert the dictionary of bind parameter values
1729 # into a dict or list to be sent to the DBAPI's
1730 # execute() or executemany() method.
1732 if compiled.positional:
1733 core_positional_parameters: MutableSequence[Sequence[Any]] = []
1734 assert positiontup is not None
1735 for compiled_params in self.compiled_parameters:
1736 l_param: List[Any] = [
1737 (
1738 flattened_processors[key](compiled_params[key])
1739 if key in flattened_processors
1740 else compiled_params[key]
1741 )
1742 for key in positiontup
1743 ]
1744 core_positional_parameters.append(
1745 dialect.execute_sequence_format(l_param)
1746 )
1748 self.parameters = core_positional_parameters
1749 else:
1750 core_dict_parameters: MutableSequence[Dict[str, Any]] = []
1751 escaped_names = compiled.escaped_bind_names
1753 # note that currently, "expanded" parameters will be present
1754 # in self.compiled_parameters in their quoted form. This is
1755 # slightly inconsistent with the approach taken as of
1756 # #8056 where self.compiled_parameters is meant to contain unquoted
1757 # param names.
1758 d_param: Dict[str, Any]
1759 for compiled_params in self.compiled_parameters:
1760 if escaped_names:
1761 d_param = {
1762 escaped_names.get(key, key): (
1763 flattened_processors[key](compiled_params[key])
1764 if key in flattened_processors
1765 else compiled_params[key]
1766 )
1767 for key in compiled_params
1768 }
1769 else:
1770 d_param = {
1771 key: (
1772 flattened_processors[key](compiled_params[key])
1773 if key in flattened_processors
1774 else compiled_params[key]
1775 )
1776 for key in compiled_params
1777 }
1779 core_dict_parameters.append(d_param)
1781 self.parameters = core_dict_parameters
1783 return self
1785 @classmethod
1786 def _init_statement(
1787 cls,
1788 dialect: Dialect,
1789 connection: Connection,
1790 dbapi_connection: PoolProxiedConnection,
1791 execution_options: _ExecuteOptions,
1792 statement: str,
1793 parameters: _DBAPIMultiExecuteParams,
1794 ) -> ExecutionContext:
1795 """Initialize execution context for a string SQL statement."""
1797 self = cls.__new__(cls)
1798 self.root_connection = connection
1799 self._dbapi_connection = dbapi_connection
1800 self.dialect = connection.dialect
1801 self.is_text = True
1803 self.execution_options = execution_options
1805 if not parameters:
1806 if self.dialect.positional:
1807 self.parameters = [dialect.execute_sequence_format()]
1808 else:
1809 self.parameters = [self._empty_dict_params]
1810 elif isinstance(parameters[0], dialect.execute_sequence_format):
1811 self.parameters = parameters
1812 elif isinstance(parameters[0], dict):
1813 self.parameters = parameters
1814 else:
1815 self.parameters = [
1816 dialect.execute_sequence_format(p) for p in parameters
1817 ]
1819 if len(parameters) > 1:
1820 self.execute_style = ExecuteStyle.EXECUTEMANY
1822 self.statement = self.unicode_statement = statement
1824 self.cursor = self.create_cursor()
1825 return self
1827 @classmethod
1828 def _init_default(
1829 cls,
1830 dialect: Dialect,
1831 connection: Connection,
1832 dbapi_connection: PoolProxiedConnection,
1833 execution_options: _ExecuteOptions,
1834 ) -> ExecutionContext:
1835 """Initialize execution context for a ColumnDefault construct."""
1837 self = cls.__new__(cls)
1838 self.root_connection = connection
1839 self._dbapi_connection = dbapi_connection
1840 self.dialect = connection.dialect
1842 self.execution_options = execution_options
1844 self.cursor = self.create_cursor()
1845 return self
1847 def _get_cache_stats(self) -> str:
1848 if self.compiled is None:
1849 return "raw sql"
1851 now = perf_counter()
1853 ch = self.cache_hit
1855 gen_time = self.compiled._gen_time
1856 assert gen_time is not None
1858 if ch is NO_CACHE_KEY:
1859 return "no key %.5fs" % (now - gen_time,)
1860 elif ch is CACHE_HIT:
1861 return "cached since %.4gs ago" % (now - gen_time,)
1862 elif ch is CACHE_MISS:
1863 return "generated in %.5fs" % (now - gen_time,)
1864 elif ch is CACHING_DISABLED:
1865 if "_cache_disable_reason" in self.execution_options:
1866 return "caching disabled (%s) %.5fs " % (
1867 self.execution_options["_cache_disable_reason"],
1868 now - gen_time,
1869 )
1870 else:
1871 return "caching disabled %.5fs" % (now - gen_time,)
1872 elif ch is NO_DIALECT_SUPPORT:
1873 return "dialect %s+%s does not support caching %.5fs" % (
1874 self.dialect.name,
1875 self.dialect.driver,
1876 now - gen_time,
1877 )
1878 else:
1879 return "unknown"
1881 @property
1882 def executemany(self): # type: ignore[override]
1883 return self.execute_style in (
1884 ExecuteStyle.EXECUTEMANY,
1885 ExecuteStyle.INSERTMANYVALUES,
1886 )
1888 @util.memoized_property
1889 def identifier_preparer(self):
1890 if self.compiled:
1891 return self.compiled.preparer
1892 elif "schema_translate_map" in self.execution_options:
1893 return self.dialect.identifier_preparer._with_schema_translate(
1894 self.execution_options["schema_translate_map"]
1895 )
1896 else:
1897 return self.dialect.identifier_preparer
1899 @util.memoized_property
1900 def engine(self):
1901 return self.root_connection.engine
1903 @util.memoized_property
1904 def postfetch_cols(self) -> Optional[Sequence[Column[Any]]]:
1905 if TYPE_CHECKING:
1906 assert isinstance(self.compiled, SQLCompiler)
1907 return self.compiled.postfetch
1909 @util.memoized_property
1910 def prefetch_cols(self) -> Optional[Sequence[Column[Any]]]:
1911 if TYPE_CHECKING:
1912 assert isinstance(self.compiled, SQLCompiler)
1913 if self.isinsert:
1914 return self.compiled.insert_prefetch
1915 elif self.isupdate:
1916 return self.compiled.update_prefetch
1917 else:
1918 return ()
1920 @util.memoized_property
1921 def no_parameters(self):
1922 return self.execution_options.get("no_parameters", False)
1924 def _execute_scalar(
1925 self,
1926 stmt: str,
1927 type_: Optional[TypeEngine[Any]],
1928 parameters: Optional[_DBAPISingleExecuteParams] = None,
1929 ) -> Any:
1930 """Execute a string statement on the current cursor, returning a
1931 scalar result.
1933 Used to fire off sequences, default phrases, and "select lastrowid"
1934 types of statements individually or in the context of a parent INSERT
1935 or UPDATE statement.
1937 """
1939 conn = self.root_connection
1941 if "schema_translate_map" in self.execution_options:
1942 schema_translate_map = self.execution_options.get(
1943 "schema_translate_map", {}
1944 )
1946 rst = self.identifier_preparer._render_schema_translates
1947 stmt = rst(stmt, schema_translate_map)
1949 if not parameters:
1950 if self.dialect.positional:
1951 parameters = self.dialect.execute_sequence_format()
1952 else:
1953 parameters = {}
1955 conn._cursor_execute(self.cursor, stmt, parameters, context=self)
1956 row = self.cursor.fetchone()
1957 if row is not None:
1958 r = row[0]
1959 else:
1960 r = None
1961 if type_ is not None:
1962 # apply type post processors to the result
1963 proc = type_._cached_result_processor(
1964 self.dialect, self.cursor.description[0][1]
1965 )
1966 if proc:
1967 return proc(r)
1968 return r
1970 @util.memoized_property
1971 def connection(self):
1972 return self.root_connection
1974 def _use_server_side_cursor(self):
1975 if not self.dialect.supports_server_side_cursors:
1976 return False
1978 if self.dialect.server_side_cursors:
1979 # this is deprecated
1980 use_server_side = self.execution_options.get(
1981 "stream_results", True
1982 ) and (
1983 self.compiled
1984 and isinstance(self.compiled.statement, expression.Selectable)
1985 or (
1986 (
1987 not self.compiled
1988 or isinstance(
1989 self.compiled.statement, expression.TextClause
1990 )
1991 )
1992 and self.unicode_statement
1993 and SERVER_SIDE_CURSOR_RE.match(self.unicode_statement)
1994 )
1995 )
1996 else:
1997 use_server_side = self.execution_options.get(
1998 "stream_results", False
1999 )
2001 return use_server_side
2003 def create_cursor(self) -> DBAPICursor:
2004 if (
2005 # inlining initial preference checks for SS cursors
2006 self.dialect.supports_server_side_cursors
2007 and (
2008 self.execution_options.get("stream_results", False)
2009 or (
2010 self.dialect.server_side_cursors
2011 and self._use_server_side_cursor()
2012 )
2013 )
2014 ):
2015 self._is_server_side = True
2016 return self.create_server_side_cursor()
2017 else:
2018 self._is_server_side = False
2019 return self.create_default_cursor()
2021 def fetchall_for_returning(self, cursor):
2022 return cursor.fetchall()
2024 def create_default_cursor(self) -> DBAPICursor:
2025 return self._dbapi_connection.cursor()
2027 def create_server_side_cursor(self) -> DBAPICursor:
2028 raise NotImplementedError()
2030 def pre_exec(self):
2031 pass
2033 def get_out_parameter_values(self, names):
2034 raise NotImplementedError(
2035 "This dialect does not support OUT parameters"
2036 )
2038 def post_exec(self):
2039 pass
2041 def get_result_processor(
2042 self, type_: TypeEngine[Any], colname: str, coltype: DBAPIType
2043 ) -> Optional[_ResultProcessorType[Any]]:
2044 """Return a 'result processor' for a given type as present in
2045 cursor.description.
2047 This has a default implementation that dialects can override
2048 for context-sensitive result type handling.
2050 """
2051 return type_._cached_result_processor(self.dialect, coltype)
2053 def get_lastrowid(self) -> int:
2054 """return self.cursor.lastrowid, or equivalent, after an INSERT.
2056 This may involve calling special cursor functions, issuing a new SELECT
2057 on the cursor (or a new one), or returning a stored value that was
2058 calculated within post_exec().
2060 This function will only be called for dialects which support "implicit"
2061 primary key generation, keep preexecute_autoincrement_sequences set to
2062 False, and when no explicit id value was bound to the statement.
2064 The function is called once for an INSERT statement that would need to
2065 return the last inserted primary key for those dialects that make use
2066 of the lastrowid concept. In these cases, it is called directly after
2067 :meth:`.ExecutionContext.post_exec`.
2069 """
2070 return self.cursor.lastrowid
2072 def handle_dbapi_exception(self, e):
2073 pass
2075 @util.non_memoized_property
2076 def rowcount(self) -> int:
2077 if self._rowcount is not None:
2078 return self._rowcount
2079 else:
2080 return self.cursor.rowcount
2082 @property
2083 def _has_rowcount(self):
2084 return self._rowcount is not None
2086 def supports_sane_rowcount(self):
2087 return self.dialect.supports_sane_rowcount
2089 def supports_sane_multi_rowcount(self):
2090 return self.dialect.supports_sane_multi_rowcount
2092 def _setup_result_proxy(self):
2093 exec_opt = self.execution_options
2095 if self._rowcount is None and exec_opt.get("preserve_rowcount", False):
2096 self._rowcount = self.cursor.rowcount
2098 yp: Optional[Union[int, bool]]
2099 if self.is_crud or self.is_text:
2100 result = self._setup_dml_or_text_result()
2101 yp = False
2102 else:
2103 yp = exec_opt.get("yield_per", None)
2104 sr = self._is_server_side or exec_opt.get("stream_results", False)
2105 strategy = self.cursor_fetch_strategy
2106 if sr and strategy is _cursor._DEFAULT_FETCH:
2107 strategy = _cursor.BufferedRowCursorFetchStrategy(
2108 self.cursor, self.execution_options
2109 )
2110 cursor_description: _DBAPICursorDescription = (
2111 strategy.alternate_cursor_description
2112 or self.cursor.description
2113 )
2114 if cursor_description is None:
2115 strategy = _cursor._NO_CURSOR_DQL
2117 result = _cursor.CursorResult(self, strategy, cursor_description)
2119 compiled = self.compiled
2121 if (
2122 compiled
2123 and not self.isddl
2124 and cast(SQLCompiler, compiled).has_out_parameters
2125 ):
2126 self._setup_out_parameters(result)
2128 self._soft_closed = result._soft_closed
2130 if yp:
2131 result = result.yield_per(yp)
2133 return result
2135 def _setup_out_parameters(self, result):
2136 compiled = cast(SQLCompiler, self.compiled)
2138 out_bindparams = [
2139 (param, name)
2140 for param, name in compiled.bind_names.items()
2141 if param.isoutparam
2142 ]
2143 out_parameters = {}
2145 for bindparam, raw_value in zip(
2146 [param for param, name in out_bindparams],
2147 self.get_out_parameter_values(
2148 [name for param, name in out_bindparams]
2149 ),
2150 ):
2151 type_ = bindparam.type
2152 impl_type = type_.dialect_impl(self.dialect)
2153 dbapi_type = impl_type.get_dbapi_type(self.dialect.loaded_dbapi)
2154 result_processor = impl_type.result_processor(
2155 self.dialect, dbapi_type
2156 )
2157 if result_processor is not None:
2158 raw_value = result_processor(raw_value)
2159 out_parameters[bindparam.key] = raw_value
2161 result.out_parameters = out_parameters
2163 def _setup_dml_or_text_result(self):
2164 compiled = cast(SQLCompiler, self.compiled)
2166 strategy: ResultFetchStrategy = self.cursor_fetch_strategy
2168 if self.isinsert:
2169 if (
2170 self.execute_style is ExecuteStyle.INSERTMANYVALUES
2171 and compiled.effective_returning
2172 ):
2173 strategy = _cursor.FullyBufferedCursorFetchStrategy(
2174 self.cursor,
2175 initial_buffer=self._insertmanyvalues_rows,
2176 # maintain alt cursor description if set by the
2177 # dialect, e.g. mssql preserves it
2178 alternate_description=(
2179 strategy.alternate_cursor_description
2180 ),
2181 )
2183 if compiled.postfetch_lastrowid:
2184 self.inserted_primary_key_rows = (
2185 self._setup_ins_pk_from_lastrowid()
2186 )
2187 # else if not self._is_implicit_returning,
2188 # the default inserted_primary_key_rows accessor will
2189 # return an "empty" primary key collection when accessed.
2191 if self._is_server_side and strategy is _cursor._DEFAULT_FETCH:
2192 strategy = _cursor.BufferedRowCursorFetchStrategy(
2193 self.cursor, self.execution_options
2194 )
2196 if strategy is _cursor._NO_CURSOR_DML:
2197 cursor_description = None
2198 else:
2199 cursor_description = (
2200 strategy.alternate_cursor_description
2201 or self.cursor.description
2202 )
2204 if cursor_description is None:
2205 strategy = _cursor._NO_CURSOR_DML
2206 elif self._num_sentinel_cols:
2207 assert self.execute_style is ExecuteStyle.INSERTMANYVALUES
2208 # the sentinel columns are handled in CursorResult._init_metadata
2209 # using essentially _reduce
2211 result: _cursor.CursorResult[Any] = _cursor.CursorResult(
2212 self, strategy, cursor_description
2213 )
2215 if self.isinsert:
2216 if self._is_implicit_returning:
2217 rows = result.all()
2219 self.returned_default_rows = rows
2221 self.inserted_primary_key_rows = (
2222 self._setup_ins_pk_from_implicit_returning(result, rows)
2223 )
2225 # test that it has a cursor metadata that is accurate. the
2226 # first row will have been fetched and current assumptions
2227 # are that the result has only one row, until executemany()
2228 # support is added here.
2229 assert result._metadata.returns_rows
2231 # Insert statement has both return_defaults() and
2232 # returning(). rewind the result on the list of rows
2233 # we just used.
2234 if self._is_supplemental_returning:
2235 result._rewind(rows)
2236 else:
2237 result._soft_close()
2238 elif not self._is_explicit_returning:
2239 result._soft_close()
2241 # we assume here the result does not return any rows.
2242 # *usually*, this will be true. However, some dialects
2243 # such as that of MSSQL/pyodbc need to SELECT a post fetch
2244 # function so this is not necessarily true.
2245 # assert not result.returns_rows
2247 elif self._is_implicit_returning:
2248 rows = result.all()
2250 if rows:
2251 self.returned_default_rows = rows
2252 self._rowcount = len(rows)
2254 if self._is_supplemental_returning:
2255 result._rewind(rows)
2256 else:
2257 result._soft_close()
2259 # test that it has a cursor metadata that is accurate.
2260 # the rows have all been fetched however.
2261 assert result._metadata.returns_rows
2263 elif not result._metadata.returns_rows:
2264 # no results, get rowcount
2265 # (which requires open cursor on some drivers)
2266 if self._rowcount is None:
2267 self._rowcount = self.cursor.rowcount
2268 result._soft_close()
2269 elif self.isupdate or self.isdelete:
2270 if self._rowcount is None:
2271 self._rowcount = self.cursor.rowcount
2272 return result
2274 @util.memoized_property
2275 def inserted_primary_key_rows(self):
2276 # if no specific "get primary key" strategy was set up
2277 # during execution, return a "default" primary key based
2278 # on what's in the compiled_parameters and nothing else.
2279 return self._setup_ins_pk_from_empty()
2281 def _setup_ins_pk_from_lastrowid(self):
2282 getter = cast(
2283 SQLCompiler, self.compiled
2284 )._inserted_primary_key_from_lastrowid_getter
2285 lastrowid = self.get_lastrowid()
2286 return [getter(lastrowid, self.compiled_parameters[0])]
2288 def _setup_ins_pk_from_empty(self):
2289 getter = cast(
2290 SQLCompiler, self.compiled
2291 )._inserted_primary_key_from_lastrowid_getter
2292 return [getter(None, param) for param in self.compiled_parameters]
2294 def _setup_ins_pk_from_implicit_returning(self, result, rows):
2295 if not rows:
2296 return []
2298 getter = cast(
2299 SQLCompiler, self.compiled
2300 )._inserted_primary_key_from_returning_getter
2301 compiled_params = self.compiled_parameters
2303 return [
2304 getter(row, param) for row, param in zip(rows, compiled_params)
2305 ]
2307 def lastrow_has_defaults(self) -> bool:
2308 return (self.isinsert or self.isupdate) and bool(
2309 cast(SQLCompiler, self.compiled).postfetch
2310 )
2312 def _prepare_set_input_sizes(
2313 self,
2314 ) -> Optional[List[Tuple[str, Any, TypeEngine[Any]]]]:
2315 """Given a cursor and ClauseParameters, prepare arguments
2316 in order to call the appropriate
2317 style of ``setinputsizes()`` on the cursor, using DB-API types
2318 from the bind parameter's ``TypeEngine`` objects.
2320 This method only called by those dialects which set the
2321 :attr:`.Dialect.bind_typing` attribute to
2322 :attr:`.BindTyping.SETINPUTSIZES`. Python-oracledb and cx_Oracle are
2323 the only DBAPIs that requires setinputsizes(); pyodbc offers it as an
2324 option.
2326 Prior to SQLAlchemy 2.0, the setinputsizes() approach was also used
2327 for pg8000 and asyncpg, which has been changed to inline rendering
2328 of casts.
2330 """
2331 if self.isddl or self.is_text:
2332 return None
2334 compiled = cast(SQLCompiler, self.compiled)
2336 inputsizes = compiled._get_set_input_sizes_lookup()
2338 if inputsizes is None:
2339 return None
2341 dialect = self.dialect
2343 # all of the rest of this... cython?
2345 if dialect._has_events:
2346 inputsizes = dict(inputsizes)
2347 dialect.dispatch.do_setinputsizes(
2348 inputsizes, self.cursor, self.statement, self.parameters, self
2349 )
2351 if compiled.escaped_bind_names:
2352 escaped_bind_names = compiled.escaped_bind_names
2353 else:
2354 escaped_bind_names = None
2356 if dialect.positional:
2357 items = [
2358 (key, compiled.binds[key])
2359 for key in compiled.positiontup or ()
2360 ]
2361 else:
2362 items = [
2363 (key, bindparam)
2364 for bindparam, key in compiled.bind_names.items()
2365 ]
2367 generic_inputsizes: List[Tuple[str, Any, TypeEngine[Any]]] = []
2368 for key, bindparam in items:
2369 if bindparam in compiled.literal_execute_params:
2370 continue
2372 if key in self._expanded_parameters:
2373 if is_tuple_type(bindparam.type):
2374 num = len(bindparam.type.types)
2375 dbtypes = inputsizes[bindparam]
2376 generic_inputsizes.extend(
2377 (
2378 (
2379 escaped_bind_names.get(paramname, paramname)
2380 if escaped_bind_names is not None
2381 else paramname
2382 ),
2383 dbtypes[idx % num],
2384 bindparam.type.types[idx % num],
2385 )
2386 for idx, paramname in enumerate(
2387 self._expanded_parameters[key]
2388 )
2389 )
2390 else:
2391 dbtype = inputsizes.get(bindparam, None)
2392 generic_inputsizes.extend(
2393 (
2394 (
2395 escaped_bind_names.get(paramname, paramname)
2396 if escaped_bind_names is not None
2397 else paramname
2398 ),
2399 dbtype,
2400 bindparam.type,
2401 )
2402 for paramname in self._expanded_parameters[key]
2403 )
2404 else:
2405 dbtype = inputsizes.get(bindparam, None)
2407 escaped_name = (
2408 escaped_bind_names.get(key, key)
2409 if escaped_bind_names is not None
2410 else key
2411 )
2413 generic_inputsizes.append(
2414 (escaped_name, dbtype, bindparam.type)
2415 )
2417 return generic_inputsizes
2419 def _exec_default(self, column, default, type_):
2420 if default.is_sequence:
2421 return self.fire_sequence(default, type_)
2422 elif default.is_callable:
2423 # this codepath is not normally used as it's inlined
2424 # into _process_execute_defaults
2425 self.current_column = column
2426 return default.arg(self)
2427 elif default.is_clause_element:
2428 return self._exec_default_clause_element(column, default, type_)
2429 else:
2430 # this codepath is not normally used as it's inlined
2431 # into _process_execute_defaults
2432 return default.arg
2434 def _exec_default_clause_element(self, column, default, type_):
2435 # execute a default that's a complete clause element. Here, we have
2436 # to re-implement a miniature version of the compile->parameters->
2437 # cursor.execute() sequence, since we don't want to modify the state
2438 # of the connection / result in progress or create new connection/
2439 # result objects etc.
2440 # .. versionchanged:: 1.4
2442 if not default._arg_is_typed:
2443 default_arg = expression.type_coerce(default.arg, type_)
2444 else:
2445 default_arg = default.arg
2446 compiled = expression.select(default_arg).compile(dialect=self.dialect)
2447 compiled_params = compiled.construct_params()
2448 processors = compiled._bind_processors
2449 if compiled.positional:
2450 parameters = self.dialect.execute_sequence_format(
2451 [
2452 (
2453 processors[key](compiled_params[key]) # type: ignore[operator] # noqa: E501
2454 if key in processors
2455 else compiled_params[key]
2456 )
2457 for key in compiled.positiontup or ()
2458 ]
2459 )
2460 else:
2461 parameters = {
2462 key: (
2463 processors[key](compiled_params[key]) # type: ignore[assignment, operator] # noqa: E501
2464 if key in processors
2465 else compiled_params[key]
2466 )
2467 for key in compiled_params
2468 }
2469 return self._execute_scalar(
2470 str(compiled), type_, parameters=parameters
2471 )
2473 current_parameters: Optional[_CoreSingleExecuteParams] = None
2474 """A dictionary of parameters applied to the current row.
2476 This attribute is only available in the context of a user-defined default
2477 generation function, e.g. as described at :ref:`context_default_functions`.
2478 It consists of a dictionary which includes entries for each column/value
2479 pair that is to be part of the INSERT or UPDATE statement. The keys of the
2480 dictionary will be the key value of each :class:`_schema.Column`,
2481 which is usually
2482 synonymous with the name.
2484 Note that the :attr:`.DefaultExecutionContext.current_parameters` attribute
2485 does not accommodate for the "multi-values" feature of the
2486 :meth:`_expression.Insert.values` method. The
2487 :meth:`.DefaultExecutionContext.get_current_parameters` method should be
2488 preferred.
2490 .. seealso::
2492 :meth:`.DefaultExecutionContext.get_current_parameters`
2494 :ref:`context_default_functions`
2496 """
2498 def get_current_parameters(self, isolate_multiinsert_groups=True):
2499 """Return a dictionary of parameters applied to the current row.
2501 This method can only be used in the context of a user-defined default
2502 generation function, e.g. as described at
2503 :ref:`context_default_functions`. When invoked, a dictionary is
2504 returned which includes entries for each column/value pair that is part
2505 of the INSERT or UPDATE statement. The keys of the dictionary will be
2506 the key value of each :class:`_schema.Column`,
2507 which is usually synonymous
2508 with the name.
2510 :param isolate_multiinsert_groups=True: indicates that multi-valued
2511 INSERT constructs created using :meth:`_expression.Insert.values`
2512 should be
2513 handled by returning only the subset of parameters that are local
2514 to the current column default invocation. When ``False``, the
2515 raw parameters of the statement are returned including the
2516 naming convention used in the case of multi-valued INSERT.
2518 .. seealso::
2520 :attr:`.DefaultExecutionContext.current_parameters`
2522 :ref:`context_default_functions`
2524 """
2525 try:
2526 parameters = self.current_parameters
2527 column = self.current_column
2528 except AttributeError:
2529 raise exc.InvalidRequestError(
2530 "get_current_parameters() can only be invoked in the "
2531 "context of a Python side column default function"
2532 )
2533 else:
2534 assert column is not None
2535 assert parameters is not None
2536 compile_state = cast(
2537 "DMLState", cast(SQLCompiler, self.compiled).compile_state
2538 )
2539 assert compile_state is not None
2540 if (
2541 isolate_multiinsert_groups
2542 and dml.isinsert(compile_state)
2543 and compile_state._has_multi_parameters
2544 ):
2545 if column._is_multiparam_column:
2546 index = column.index + 1
2547 d = {column.original.key: parameters[column.key]}
2548 else:
2549 d = {column.key: parameters[column.key]}
2550 index = 0
2551 assert compile_state._dict_parameters is not None
2552 keys = compile_state._dict_parameters.keys()
2553 d.update(
2554 (key, parameters["%s_m%d" % (key, index)]) for key in keys
2555 )
2556 return d
2557 else:
2558 return parameters
2560 def get_insert_default(self, column):
2561 if column.default is None:
2562 return None
2563 else:
2564 return self._exec_default(column, column.default, column.type)
2566 def get_update_default(self, column):
2567 if column.onupdate is None:
2568 return None
2569 else:
2570 return self._exec_default(column, column.onupdate, column.type)
2572 def _process_execute_defaults(self):
2573 compiled = cast(SQLCompiler, self.compiled)
2575 key_getter = compiled._within_exec_param_key_getter
2577 sentinel_counter = 0
2579 if compiled.insert_prefetch:
2580 prefetch_recs = [
2581 (
2582 c,
2583 key_getter(c),
2584 c._default_description_tuple,
2585 self.get_insert_default,
2586 )
2587 for c in compiled.insert_prefetch
2588 ]
2589 elif compiled.update_prefetch:
2590 prefetch_recs = [
2591 (
2592 c,
2593 key_getter(c),
2594 c._onupdate_description_tuple,
2595 self.get_update_default,
2596 )
2597 for c in compiled.update_prefetch
2598 ]
2599 else:
2600 prefetch_recs = []
2602 for param in self.compiled_parameters:
2603 self.current_parameters = param
2605 for (
2606 c,
2607 param_key,
2608 (arg, is_scalar, is_callable, is_sentinel),
2609 fallback,
2610 ) in prefetch_recs:
2611 if is_sentinel:
2612 param[param_key] = sentinel_counter
2613 sentinel_counter += 1
2614 elif is_scalar:
2615 param[param_key] = arg
2616 elif is_callable:
2617 self.current_column = c
2618 param[param_key] = arg(self)
2619 else:
2620 val = fallback(c)
2621 if val is not None:
2622 param[param_key] = val
2624 del self.current_parameters
2627DefaultDialect.execution_ctx_cls = DefaultExecutionContext