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 datetime
20import functools
21import operator
22import random
23import re
24from time import perf_counter
25import typing
26from typing import Any
27from typing import Callable
28from typing import cast
29from typing import Dict
30from typing import Final
31from typing import Iterable
32from typing import List
33from typing import Literal
34from typing import Mapping
35from typing import MutableMapping
36from typing import MutableSequence
37from typing import Optional
38from typing import Sequence
39from typing import Set
40from typing import Tuple
41from typing import Type
42from typing import TYPE_CHECKING
43from typing import Union
44import weakref
46from . import characteristics
47from . import cursor as _cursor
48from . import interfaces
49from . import reflection
50from .base import Connection
51from .interfaces import CacheStats
52from .interfaces import DBAPICursor
53from .interfaces import Dialect
54from .interfaces import ExecuteStyle
55from .interfaces import ExecutionContext
56from .reflection import ObjectKind
57from .reflection import ObjectScope
58from .. import event
59from .. import exc
60from .. import pool
61from .. import util
62from ..sql import compiler
63from ..sql import dml
64from ..sql import expression
65from ..sql import type_api
66from ..sql import util as sql_util
67from ..sql._typing import is_tuple_type
68from ..sql.base import _NoArg
69from ..sql.compiler import AggregateOrderByStyle
70from ..sql.compiler import DDLCompiler
71from ..sql.compiler import InsertmanyvaluesSentinelOpts
72from ..sql.compiler import SQLCompiler
73from ..sql.elements import quoted_name
74from ..util.typing import TupleAny
75from ..util.typing import Unpack
77if typing.TYPE_CHECKING:
78 from .base import Engine
79 from .cursor import ResultFetchStrategy
80 from .interfaces import _CoreMultiExecuteParams
81 from .interfaces import _CoreSingleExecuteParams
82 from .interfaces import _DBAPICursorDescription
83 from .interfaces import _DBAPIMultiExecuteParams
84 from .interfaces import _DBAPISingleExecuteParams
85 from .interfaces import _ExecuteOptions
86 from .interfaces import _MutableCoreSingleExecuteParams
87 from .interfaces import _ParamStyle
88 from .interfaces import ConnectArgsType
89 from .interfaces import DBAPIConnection
90 from .interfaces import DBAPIModule
91 from .interfaces import DBAPIType
92 from .interfaces import IsolationLevel
93 from .interfaces import TableKey
94 from .row import Row
95 from .url import URL
96 from ..event import _ListenerFnType
97 from ..pool import Pool
98 from ..pool import PoolProxiedConnection
99 from ..sql import Executable
100 from ..sql.compiler import Compiled
101 from ..sql.compiler import Linting
102 from ..sql.compiler import ResultColumnsEntry
103 from ..sql.dml import DMLState
104 from ..sql.dml import UpdateBase
105 from ..sql.elements import BindParameter
106 from ..sql.schema import Column
107 from ..sql.sqltypes import _JSON_VALUE
108 from ..sql.type_api import _BindProcessorType
109 from ..sql.type_api import _ResultProcessorType
110 from ..sql.type_api import TypeEngine
113# When we're handed literal SQL, ensure it's a SELECT query
114SERVER_SIDE_CURSOR_RE = re.compile(r"\s*SELECT", re.I | re.UNICODE)
117(
118 CACHE_HIT,
119 CACHE_MISS,
120 CACHING_DISABLED,
121 NO_CACHE_KEY,
122 NO_DIALECT_SUPPORT,
123) = list(CacheStats)
126class _BackendsMultiReflection(Dialect):
127 """Mixin providing single-table reflection wrappers that delegate to
128 the corresponding ``get_multi_*`` methods.
130 Used by dialects that implement native multi-table reflection
131 (PostgreSQL, Oracle, MSSQL).
132 """
134 @reflection.cache
135 def has_table(
136 self,
137 connection: Connection,
138 table_name: str,
139 schema: Optional[str] = None,
140 **kw: Any,
141 ) -> bool:
142 # NOTE: assume it's a subclass of DefaultDialect
143 self._ensure_has_table_connection(connection) # type: ignore[attr-defined] # noqa: E501
144 multi_res = self.has_multi_table(
145 connection,
146 table_names=[table_name],
147 schema=schema,
148 **kw,
149 )
150 # has_multi_table returns all the input table names so it's not
151 # possible for the key to be missing
152 return dict(multi_res)[(schema, table_name)]
154 def _value_or_raise(self, data, table, schema):
155 try:
156 return dict(data)[(schema, table)]
157 except KeyError:
158 raise exc.NoSuchTableError(
159 f"{schema}.{table}" if schema else table
160 ) from None
162 @reflection.cache
163 def get_columns(self, connection, table_name, schema=None, **kw):
164 data = self.get_multi_columns(
165 connection,
166 schema=schema,
167 filter_names=[table_name],
168 scope=ObjectScope.ANY,
169 kind=ObjectKind.ANY,
170 **kw,
171 )
172 return self._value_or_raise(data, table_name, schema)
174 @reflection.cache
175 def get_table_options(self, connection, table_name, schema=None, **kw):
176 data = self.get_multi_table_options(
177 connection,
178 schema=schema,
179 filter_names=[table_name],
180 scope=ObjectScope.ANY,
181 kind=ObjectKind.ANY,
182 **kw,
183 )
184 return self._value_or_raise(data, table_name, schema)
186 @reflection.cache
187 def get_pk_constraint(self, connection, table_name, schema=None, **kw):
188 data = self.get_multi_pk_constraint(
189 connection,
190 schema=schema,
191 filter_names=[table_name],
192 scope=ObjectScope.ANY,
193 kind=ObjectKind.ANY,
194 **kw,
195 )
196 return self._value_or_raise(data, table_name, schema)
198 @reflection.cache
199 def get_foreign_keys(self, connection, table_name, schema=None, **kw):
200 data = self.get_multi_foreign_keys(
201 connection,
202 schema=schema,
203 filter_names=[table_name],
204 scope=ObjectScope.ANY,
205 kind=ObjectKind.ANY,
206 **kw,
207 )
208 return self._value_or_raise(data, table_name, schema)
210 @reflection.cache
211 def get_indexes(self, connection, table_name, schema=None, **kw):
212 data = self.get_multi_indexes(
213 connection,
214 schema=schema,
215 filter_names=[table_name],
216 scope=ObjectScope.ANY,
217 kind=ObjectKind.ANY,
218 **kw,
219 )
220 return self._value_or_raise(data, table_name, schema)
222 @reflection.cache
223 def get_unique_constraints(
224 self, connection, table_name, schema=None, **kw
225 ):
226 data = self.get_multi_unique_constraints(
227 connection,
228 schema=schema,
229 filter_names=[table_name],
230 scope=ObjectScope.ANY,
231 kind=ObjectKind.ANY,
232 **kw,
233 )
234 return self._value_or_raise(data, table_name, schema)
236 @reflection.cache
237 def get_check_constraints(self, connection, table_name, schema=None, **kw):
238 data = self.get_multi_check_constraints(
239 connection,
240 schema=schema,
241 filter_names=[table_name],
242 scope=ObjectScope.ANY,
243 kind=ObjectKind.ANY,
244 **kw,
245 )
246 return self._value_or_raise(data, table_name, schema)
248 @reflection.cache
249 def get_table_comment(self, connection, table_name, schema=None, **kw):
250 data = self.get_multi_table_comment(
251 connection,
252 schema=schema,
253 filter_names=[table_name],
254 scope=ObjectScope.ANY,
255 kind=ObjectKind.ANY,
256 **kw,
257 )
258 return self._value_or_raise(data, table_name, schema)
261class DefaultDialect(Dialect):
262 """Default implementation of Dialect"""
264 statement_compiler = compiler.SQLCompiler
265 ddl_compiler = compiler.DDLCompiler
266 type_compiler_cls = compiler.GenericTypeCompiler
268 preparer = compiler.IdentifierPreparer
269 supports_alter = True
270 supports_comments = False
271 supports_constraint_comments = False
272 inline_comments = False
273 supports_statement_cache = True
275 div_is_floordiv = True
277 bind_typing = interfaces.BindTyping.NONE
279 include_set_input_sizes: Optional[Set[Any]] = None
280 exclude_set_input_sizes: Optional[Set[Any]] = None
282 # the first value we'd get for an autoincrement column.
283 default_sequence_base = 1
285 # most DBAPIs happy with this for execute().
286 # not cx_oracle.
287 execute_sequence_format = tuple
289 supports_schemas = True
290 supports_views = True
291 supports_sequences = False
292 sequences_optional = False
293 preexecute_autoincrement_sequences = False
294 supports_identity_columns = False
295 postfetch_lastrowid = True
296 favor_returning_over_lastrowid = False
297 insert_null_pk_still_autoincrements = False
298 update_returning = False
299 delete_returning = False
300 update_returning_multifrom = False
301 delete_returning_multifrom = False
302 insert_returning = False
304 aggregate_order_by_style = AggregateOrderByStyle.INLINE
306 cte_follows_insert = False
308 supports_native_enum = False
309 supports_native_boolean = False
310 supports_native_uuid = False
311 returns_native_bytes = False
313 supports_native_json_serialization = False
314 supports_native_json_deserialization = False
315 dialect_injects_custom_json_deserializer = False
316 _json_serializer: Callable[[_JSON_VALUE], str] | None = None
318 _json_deserializer: Callable[[str], _JSON_VALUE] | None = None
320 non_native_boolean_check_constraint = True
322 supports_simple_order_by_label = True
324 tuple_in_values = False
326 connection_characteristics = util.immutabledict(
327 {
328 "isolation_level": characteristics.IsolationLevelCharacteristic(),
329 "logging_token": characteristics.LoggingTokenCharacteristic(),
330 }
331 )
333 engine_config_types: Mapping[str, Any] = util.immutabledict(
334 {
335 "pool_timeout": util.asint,
336 "echo": util.bool_or_str("debug"),
337 "echo_pool": util.bool_or_str("debug"),
338 "pool_recycle": util.asint,
339 "pool_size": util.asint,
340 "max_overflow": util.asint,
341 "future": util.asbool,
342 }
343 )
345 # if the NUMERIC type
346 # returns decimal.Decimal.
347 # *not* the FLOAT type however.
348 supports_native_decimal = False
350 name = "default"
352 # length at which to truncate
353 # any identifier.
354 max_identifier_length = 9999
355 _user_defined_max_identifier_length: Optional[int] = None
357 isolation_level: Optional[str] = None
359 # sub-categories of max_identifier_length.
360 # currently these accommodate for MySQL which allows alias names
361 # of 255 but DDL names only of 64.
362 max_index_name_length: Optional[int] = None
363 max_constraint_name_length: Optional[int] = None
365 supports_sane_rowcount = True
366 supports_sane_multi_rowcount = True
367 colspecs: MutableMapping[Type[TypeEngine[Any]], Type[TypeEngine[Any]]] = {}
368 default_paramstyle = "named"
370 supports_default_values = False
371 """dialect supports INSERT... DEFAULT VALUES syntax"""
373 supports_default_metavalue = False
374 """dialect supports INSERT... VALUES (DEFAULT) syntax"""
376 default_metavalue_token = "DEFAULT"
377 """for INSERT... VALUES (DEFAULT) syntax, the token to put in the
378 parenthesis."""
380 # not sure if this is a real thing but the compiler will deliver it
381 # if this is the only flag enabled.
382 supports_empty_insert = True
383 """dialect supports INSERT () VALUES ()"""
385 supports_multivalues_insert = False
387 use_insertmanyvalues: bool = False
389 use_insertmanyvalues_wo_returning: bool = False
391 insertmanyvalues_implicit_sentinel: InsertmanyvaluesSentinelOpts = (
392 InsertmanyvaluesSentinelOpts.NOT_SUPPORTED
393 )
395 insertmanyvalues_page_size: int = 1000
396 insertmanyvalues_max_parameters = 32700
398 supports_is_distinct_from = True
400 supports_server_side_cursors = False
402 server_side_cursors = False
404 # extra record-level locking features (#4860)
405 supports_for_update_of = False
407 server_version_info = None
409 default_schema_name: Optional[str] = None
411 # indicates symbol names are
412 # UPPERCASED if they are case insensitive
413 # within the database.
414 # if this is True, the methods normalize_name()
415 # and denormalize_name() must be provided.
416 requires_name_normalize = False
418 is_async = False
420 has_terminate = False
422 # TODO: this is not to be part of 2.0. implement rudimentary binary
423 # literals for SQLite, PostgreSQL, MySQL only within
424 # _Binary.literal_processor
425 _legacy_binary_type_literal_encoding = "utf-8"
427 @util.deprecated_params(
428 empty_in_strategy=(
429 "1.4",
430 "The :paramref:`_sa.create_engine.empty_in_strategy` keyword is "
431 "deprecated, and no longer has any effect. All IN expressions "
432 "are now rendered using "
433 'the "expanding parameter" strategy which renders a set of bound'
434 'expressions, or an "empty set" SELECT, at statement execution'
435 "time.",
436 ),
437 server_side_cursors=(
438 "1.4",
439 "The :paramref:`_sa.create_engine.server_side_cursors` parameter "
440 "is deprecated and will be removed in a future release. Please "
441 "use the "
442 ":paramref:`_engine.Connection.execution_options.stream_results` "
443 "parameter.",
444 ),
445 )
446 def __init__(
447 self,
448 paramstyle: Optional[_ParamStyle] = None,
449 isolation_level: Optional[IsolationLevel] = None,
450 dbapi: Optional[DBAPIModule] = None,
451 implicit_returning: Literal[True] = True,
452 supports_native_boolean: Optional[bool] = None,
453 max_identifier_length: Optional[int] = None,
454 label_length: Optional[int] = None,
455 insertmanyvalues_page_size: Union[_NoArg, int] = _NoArg.NO_ARG,
456 use_insertmanyvalues: Optional[bool] = None,
457 # util.deprecated_params decorator cannot render the
458 # Linting.NO_LINTING constant
459 compiler_linting: Linting = int(compiler.NO_LINTING), # type: ignore[assignment] # noqa: E501
460 server_side_cursors: bool = False,
461 skip_autocommit_rollback: bool = False,
462 **kwargs: Any,
463 ):
464 if server_side_cursors:
465 if not self.supports_server_side_cursors:
466 raise exc.ArgumentError(
467 "Dialect %s does not support server side cursors" % self
468 )
469 else:
470 self.server_side_cursors = True
472 if getattr(self, "use_setinputsizes", False):
473 util.warn_deprecated(
474 "The dialect-level use_setinputsizes attribute is "
475 "deprecated. Please use "
476 "bind_typing = BindTyping.SETINPUTSIZES",
477 "2.0",
478 )
479 self.bind_typing = interfaces.BindTyping.SETINPUTSIZES
481 self.positional = False
482 self._ischema = None
484 self.dbapi = dbapi
486 self.skip_autocommit_rollback = skip_autocommit_rollback
488 if paramstyle is not None:
489 self.paramstyle = paramstyle
490 elif self.dbapi is not None:
491 self.paramstyle = self.dbapi.paramstyle
492 else:
493 self.paramstyle = self.default_paramstyle
494 self.positional = self.paramstyle in (
495 "qmark",
496 "format",
497 "numeric",
498 "numeric_dollar",
499 )
500 self.identifier_preparer = self.preparer(self)
501 self._on_connect_isolation_level = isolation_level
503 legacy_tt_callable = getattr(self, "type_compiler", None)
504 if legacy_tt_callable is not None:
505 tt_callable = cast(
506 Type[compiler.GenericTypeCompiler],
507 self.type_compiler,
508 )
509 else:
510 tt_callable = self.type_compiler_cls
512 self.type_compiler_instance = self.type_compiler = tt_callable(self)
514 if supports_native_boolean is not None:
515 self.supports_native_boolean = supports_native_boolean
517 self._user_defined_max_identifier_length = max_identifier_length
518 if self._user_defined_max_identifier_length:
519 self.max_identifier_length = (
520 self._user_defined_max_identifier_length
521 )
522 self.label_length = label_length
523 self.compiler_linting = compiler_linting
525 if use_insertmanyvalues is not None:
526 self.use_insertmanyvalues = use_insertmanyvalues
528 if insertmanyvalues_page_size is not _NoArg.NO_ARG:
529 self.insertmanyvalues_page_size = insertmanyvalues_page_size
531 self._check_minimum_dbapi_version()
533 @property
534 @util.deprecated(
535 "2.0",
536 "full_returning is deprecated, please use insert_returning, "
537 "update_returning, delete_returning",
538 )
539 def full_returning(self):
540 return (
541 self.insert_returning
542 and self.update_returning
543 and self.delete_returning
544 )
546 @util.memoized_property
547 def insert_executemany_returning(self):
548 """Default implementation for insert_executemany_returning, if not
549 otherwise overridden by the specific dialect.
551 The default dialect determines "insert_executemany_returning" is
552 available if the dialect in use has opted into using the
553 "use_insertmanyvalues" feature. If they haven't opted into that, then
554 this attribute is False, unless the dialect in question overrides this
555 and provides some other implementation (such as the Oracle Database
556 dialects).
558 """
559 return self.insert_returning and self.use_insertmanyvalues
561 @util.memoized_property
562 def insert_executemany_returning_sort_by_parameter_order(self):
563 """Default implementation for
564 insert_executemany_returning_deterministic_order, if not otherwise
565 overridden by the specific dialect.
567 The default dialect determines "insert_executemany_returning" can have
568 deterministic order only if the dialect in use has opted into using the
569 "use_insertmanyvalues" feature, which implements deterministic ordering
570 using client side sentinel columns only by default. The
571 "insertmanyvalues" feature also features alternate forms that can
572 use server-generated PK values as "sentinels", but those are only
573 used if the :attr:`.Dialect.insertmanyvalues_implicit_sentinel`
574 bitflag enables those alternate SQL forms, which are disabled
575 by default.
577 If the dialect in use hasn't opted into that, then this attribute is
578 False, unless the dialect in question overrides this and provides some
579 other implementation (such as the Oracle Database dialects).
581 """
582 return self.insert_returning and self.use_insertmanyvalues
584 update_executemany_returning = False
585 delete_executemany_returning = False
587 @util.memoized_property
588 def loaded_dbapi(self) -> DBAPIModule:
589 if self.dbapi is None:
590 raise exc.NoDBAPILoaded(
591 f"Dialect {self} does not have a Python DBAPI established "
592 "and cannot be used for actual database interaction"
593 )
594 return self.dbapi
596 @util.memoized_property
597 def dbapi_version(self) -> util.VersionInfo:
598 # memoization applies to a successfully determined version only;
599 # memoized_property does not cache when the function raises, so a
600 # DBAPI which is established after this dialect was constructed is
601 # still picked up
602 if self.dbapi is None:
603 raise exc.NoDBAPILoaded(
604 f"Dialect {self.name}+{self.driver} has no DBAPI module "
605 "loaded; no DBAPI version is available"
606 )
608 try:
609 version = self.retrieve_dbapi_version(self.dbapi)
610 except NotImplementedError as ne:
611 raise NotImplementedError(
612 f"Dialect {self.name}+{self.driver} does not implement "
613 "retrieve_dbapi_version(); no DBAPI version is available"
614 ) from ne
616 if not version:
617 # the DBAPI is loaded but publishes no version of its own; as
618 # with a DBAPI that isn't loaded, this is not an error on the
619 # part of the dialect
620 # asyncio dialects have a wrapper object here rather than a
621 # module, which has no __name__
622 dbapi_name = getattr(self.dbapi, "__name__", self.driver)
623 raise exc.NoDBAPILoaded(
624 f"Dialect {self.name}+{self.driver} could not determine a "
625 f"version for its DBAPI module {dbapi_name!r}"
626 )
628 return version
630 @property
631 def _dbapi_version_or_none(self) -> Optional[util.VersionInfo]:
632 """:attr:`.Dialect.dbapi_version`, or None if it can't be
633 determined.
635 For use by dialect startup checks, which run before a DBAPI is
636 necessarily present and which must not fail when no version is
637 available. Only :class:`.exc.NoDBAPILoaded` is accommodated;
638 ``NotImplementedError``, indicating a dialect which does not
639 implement :meth:`.Dialect.retrieve_dbapi_version` at all, is a bug
640 in that dialect and is allowed to propagate. Deliberately not
641 memoized itself; the underlying :attr:`.Dialect.dbapi_version`
642 memoizes the success case.
644 """
645 try:
646 return self.dbapi_version
647 except exc.NoDBAPILoaded:
648 return None
650 def _check_minimum_dbapi_version(self) -> None:
651 """Enforce :attr:`.Dialect.minimum_dbapi_version`, if present.
653 Takes place as the dialect is constructed. No check occurs when
654 the version of the DBAPI is not available at all.
656 """
657 minimum = self.minimum_dbapi_version
658 if minimum is None:
659 return
661 version = self._dbapi_version_or_none
662 if version is not None and version < minimum:
663 dbapi_name = getattr(self.dbapi, "__name__", self.driver)
664 raise exc.InvalidRequestError(
665 f"Dialect {self.name}+{self.driver} requires version "
666 f"{minimum} or greater of the {dbapi_name} DBAPI; "
667 f"version {version} is installed"
668 )
670 @util.memoized_property
671 def _bind_typing_render_casts(self):
672 return self.bind_typing is interfaces.BindTyping.RENDER_CASTS
674 def _ensure_has_table_connection(self, arg: Connection) -> None:
675 if not isinstance(arg, Connection):
676 raise exc.ArgumentError(
677 "The argument passed to Dialect.has_table() should be a "
678 "%s, got %s. "
679 "Additionally, the Dialect.has_table() method is for "
680 "internal dialect "
681 "use only; please use "
682 "``inspect(some_engine).has_table(<tablename>>)`` "
683 "for public API use." % (Connection, type(arg))
684 )
686 @util.memoized_property
687 def _supports_statement_cache(self):
688 ssc = self.__class__.__dict__.get("supports_statement_cache", None)
689 if ssc is None:
690 util.warn(
691 "Dialect %s:%s will not make use of SQL compilation caching "
692 "as it does not set the 'supports_statement_cache' attribute "
693 "to ``True``. This can have "
694 "significant performance implications including some "
695 "performance degradations in comparison to prior SQLAlchemy "
696 "versions. Dialect maintainers should seek to set this "
697 "attribute to True after appropriate development and testing "
698 "for SQLAlchemy 1.4 caching support. Alternatively, this "
699 "attribute may be set to False which will disable this "
700 "warning." % (self.name, self.driver),
701 code="cprf",
702 )
704 return bool(ssc)
706 @util.memoized_property
707 def _type_memos(self):
708 return weakref.WeakKeyDictionary()
710 @property
711 def dialect_description(self): # type: ignore[override]
712 return self.name + "+" + self.driver
714 @property
715 def supports_sane_rowcount_returning(self):
716 """True if this dialect supports sane rowcount even if RETURNING is
717 in use.
719 For dialects that don't support RETURNING, this is synonymous with
720 ``supports_sane_rowcount``.
722 """
723 return self.supports_sane_rowcount
725 @classmethod
726 def get_pool_class(cls, url: URL) -> Type[Pool]:
727 default: Type[pool.Pool]
728 if cls.is_async:
729 default = pool.AsyncAdaptedQueuePool
730 else:
731 default = pool.QueuePool
733 return getattr(cls, "poolclass", default)
735 def get_dialect_pool_class(self, url: URL) -> Type[Pool]:
736 return self.get_pool_class(url)
738 @classmethod
739 def load_provisioning(cls):
740 package = ".".join(cls.__module__.split(".")[0:-1])
741 try:
742 __import__(package + ".provision")
743 except ImportError:
744 pass
746 def _builtin_onconnect(self) -> Optional[_ListenerFnType]:
747 if self._on_connect_isolation_level is not None:
749 def builtin_connect(dbapi_conn, conn_rec):
750 self._assert_and_set_isolation_level(
751 dbapi_conn, self._on_connect_isolation_level
752 )
754 return builtin_connect
755 else:
756 return None
758 def initialize(self, connection: Connection) -> None:
759 try:
760 self.server_version_info = self._get_server_version_info(
761 connection
762 )
763 except NotImplementedError:
764 self.server_version_info = None
765 try:
766 self.default_schema_name = self._get_default_schema_name(
767 connection
768 )
769 except NotImplementedError:
770 self.default_schema_name = None
772 try:
773 self.default_isolation_level = self.get_default_isolation_level(
774 connection.connection.dbapi_connection
775 )
776 except NotImplementedError:
777 self.default_isolation_level = None
779 if not self._user_defined_max_identifier_length:
780 max_ident_length = self._check_max_identifier_length(connection)
781 if max_ident_length:
782 self.max_identifier_length = max_ident_length
784 if (
785 self.label_length
786 and self.label_length > self.max_identifier_length
787 ):
788 raise exc.ArgumentError(
789 "Label length of %d is greater than this dialect's"
790 " maximum identifier length of %d"
791 % (self.label_length, self.max_identifier_length)
792 )
794 def on_connect(self) -> Optional[Callable[[Any], None]]:
795 # inherits the docstring from interfaces.Dialect.on_connect
796 return None
798 def _check_max_identifier_length(self, connection):
799 """Perform a connection / server version specific check to determine
800 the max_identifier_length.
802 If the dialect's class level max_identifier_length should be used,
803 can return None.
805 """
806 return None
808 def get_default_isolation_level(self, dbapi_conn):
809 """Given a DBAPI connection, return its isolation level, or
810 a default isolation level if one cannot be retrieved.
812 May be overridden by subclasses in order to provide a
813 "fallback" isolation level for databases that cannot reliably
814 retrieve the actual isolation level.
816 By default, calls the :meth:`_engine.Interfaces.get_isolation_level`
817 method, propagating any exceptions raised.
819 """
820 return self.get_isolation_level(dbapi_conn)
822 def type_descriptor(self, typeobj):
823 """Provide a database-specific :class:`.TypeEngine` object, given
824 the generic object which comes from the types module.
826 This method looks for a dictionary called
827 ``colspecs`` as a class or instance-level variable,
828 and passes on to :func:`_types.adapt_type`.
830 """
831 return type_api.adapt_type(typeobj, self.colspecs)
833 def min_year_for_type(self, typeobj: TypeEngine[Any]) -> int:
834 """Return the earliest year the database accepts for the given
835 date or datetime type.
837 The default implementation returns ``datetime.MINYEAR``.
839 .. versionadded:: 2.0.55
841 """
842 return datetime.MINYEAR
844 def has_index(self, connection, table_name, index_name, schema=None, **kw):
845 if not self.has_table(connection, table_name, schema=schema, **kw):
846 return False
847 for idx in self.get_indexes(
848 connection, table_name, schema=schema, **kw
849 ):
850 if idx["name"] == index_name:
851 return True
852 else:
853 return False
855 def has_schema(
856 self, connection: Connection, schema_name: str, **kw: Any
857 ) -> bool:
858 return schema_name in self.get_schema_names(connection, **kw)
860 def validate_identifier(self, ident: str) -> None:
861 if len(ident) > self.max_identifier_length:
862 raise exc.IdentifierError(
863 "Identifier '%s' exceeds maximum length of %d characters"
864 % (ident, self.max_identifier_length)
865 )
867 def connect(self, *cargs: Any, **cparams: Any) -> DBAPIConnection:
868 # inherits the docstring from interfaces.Dialect.connect
869 return self.loaded_dbapi.connect(*cargs, **cparams) # type: ignore[no-any-return] # NOQA: E501
871 def create_connect_args(self, url: URL) -> ConnectArgsType:
872 # inherits the docstring from interfaces.Dialect.create_connect_args
873 opts = url.translate_connect_args()
874 opts.update(url.query)
875 return ([], opts)
877 def set_engine_execution_options(
878 self, engine: Engine, opts: Mapping[str, Any]
879 ) -> None:
880 supported_names = set(self.connection_characteristics).intersection(
881 opts
882 )
883 if supported_names:
884 characteristics: Mapping[str, Any] = util.immutabledict(
885 (name, opts[name]) for name in supported_names
886 )
888 @event.listens_for(engine, "engine_connect")
889 def set_connection_characteristics(connection):
890 self._set_connection_characteristics(
891 connection, characteristics
892 )
894 def set_connection_execution_options(
895 self, connection: Connection, opts: Mapping[str, Any]
896 ) -> None:
897 supported_names = set(self.connection_characteristics).intersection(
898 opts
899 )
900 if supported_names:
901 characteristics: Mapping[str, Any] = util.immutabledict(
902 (name, opts[name]) for name in supported_names
903 )
904 self._set_connection_characteristics(connection, characteristics)
906 def _set_connection_characteristics(self, connection, characteristics):
907 characteristic_values = [
908 (name, self.connection_characteristics[name], value)
909 for name, value in characteristics.items()
910 ]
912 if connection.in_transaction():
913 trans_objs = [
914 (name, obj)
915 for name, obj, _ in characteristic_values
916 if obj.transactional
917 ]
918 if trans_objs:
919 raise exc.InvalidRequestError(
920 "This connection has already initialized a SQLAlchemy "
921 "Transaction() object via begin() or autobegin; "
922 "%s may not be altered unless rollback() or commit() "
923 "is called first."
924 % (", ".join(name for name, obj in trans_objs))
925 )
927 dbapi_connection = connection.connection.dbapi_connection
928 for _, characteristic, value in characteristic_values:
929 characteristic.set_connection_characteristic(
930 self, connection, dbapi_connection, value
931 )
932 connection.connection._connection_record.finalize_callback.append(
933 functools.partial(self._reset_characteristics, characteristics)
934 )
936 def _reset_characteristics(self, characteristics, dbapi_connection):
937 for characteristic_name in characteristics:
938 characteristic = self.connection_characteristics[
939 characteristic_name
940 ]
941 characteristic.reset_characteristic(self, dbapi_connection)
943 def do_begin(self, dbapi_connection):
944 pass
946 def do_rollback(self, dbapi_connection):
947 if self.skip_autocommit_rollback and self.detect_autocommit_setting(
948 dbapi_connection
949 ):
950 return
951 dbapi_connection.rollback()
953 def do_commit(self, dbapi_connection):
954 dbapi_connection.commit()
956 def do_terminate(self, dbapi_connection):
957 self.do_close(dbapi_connection)
959 def do_close(self, dbapi_connection):
960 dbapi_connection.close()
962 @util.memoized_property
963 def _dialect_specific_select_one(self):
964 return str(expression.select(1).compile(dialect=self))
966 def _do_ping_w_event(self, dbapi_connection: DBAPIConnection) -> bool:
967 try:
968 return self.do_ping(dbapi_connection)
969 except self.loaded_dbapi.Error as err:
970 is_disconnect = self.is_disconnect(err, dbapi_connection, None)
972 if self._has_events:
973 try:
974 Connection._handle_dbapi_exception_noconnection(
975 err,
976 self,
977 is_disconnect=is_disconnect,
978 invalidate_pool_on_disconnect=False,
979 is_pre_ping=True,
980 )
981 except exc.StatementError as new_err:
982 is_disconnect = new_err.connection_invalidated
984 if is_disconnect:
985 return False
986 else:
987 raise
989 def do_ping(self, dbapi_connection: DBAPIConnection) -> bool:
990 cursor = dbapi_connection.cursor()
991 try:
992 cursor.execute(self._dialect_specific_select_one)
993 finally:
994 cursor.close()
995 return True
997 def create_xid(self):
998 """Create a random two-phase transaction ID.
1000 This id will be passed to do_begin_twophase(), do_rollback_twophase(),
1001 do_commit_twophase(). Its format is unspecified.
1002 """
1004 return "_sa_%032x" % random.randint(0, 2**128)
1006 def do_savepoint(self, connection, name):
1007 connection.execute(expression.SavepointClause(name))
1009 def do_rollback_to_savepoint(self, connection, name):
1010 connection.execute(expression.RollbackToSavepointClause(name))
1012 def do_release_savepoint(self, connection, name):
1013 connection.execute(expression.ReleaseSavepointClause(name))
1015 def _deliver_insertmanyvalues_batches(
1016 self,
1017 connection,
1018 cursor,
1019 statement,
1020 parameters,
1021 generic_setinputsizes,
1022 context,
1023 ):
1024 context = cast(DefaultExecutionContext, context)
1025 compiled = cast(SQLCompiler, context.compiled)
1027 _composite_sentinel_proc: Sequence[
1028 Optional[_ResultProcessorType[Any]]
1029 ] = ()
1030 _scalar_sentinel_proc: Optional[_ResultProcessorType[Any]] = None
1031 _sentinel_proc_initialized: bool = False
1033 compiled_parameters = context.compiled_parameters
1035 imv = compiled._insertmanyvalues
1036 assert imv is not None
1038 is_returning: Final[bool] = bool(compiled.effective_returning)
1039 batch_size = context.execution_options.get(
1040 "insertmanyvalues_page_size", self.insertmanyvalues_page_size
1041 )
1043 if compiled.schema_translate_map:
1044 schema_translate_map = context.execution_options.get(
1045 "schema_translate_map", {}
1046 )
1047 else:
1048 schema_translate_map = None
1050 if is_returning:
1051 result: Optional[List[Any]] = []
1052 context._insertmanyvalues_rows = result
1054 sort_by_parameter_order = imv.sort_by_parameter_order
1056 else:
1057 sort_by_parameter_order = False
1058 result = None
1060 for imv_batch in compiled._deliver_insertmanyvalues_batches(
1061 statement,
1062 parameters,
1063 compiled_parameters,
1064 generic_setinputsizes,
1065 batch_size,
1066 sort_by_parameter_order,
1067 schema_translate_map,
1068 ):
1069 yield imv_batch
1071 if is_returning:
1073 try:
1074 rows = context.fetchall_for_returning(cursor)
1075 except BaseException as be:
1076 connection._handle_dbapi_exception(
1077 be,
1078 sql_util._long_statement(imv_batch.replaced_statement),
1079 imv_batch.replaced_parameters,
1080 None,
1081 context,
1082 is_sub_exec=True,
1083 )
1085 # I would have thought "is_returning: Final[bool]"
1086 # would have assured this but pylance thinks not
1087 assert result is not None
1089 if imv.num_sentinel_columns and not imv_batch.is_downgraded:
1090 composite_sentinel = imv.num_sentinel_columns > 1
1091 if imv.implicit_sentinel:
1092 # for implicit sentinel, which is currently single-col
1093 # integer autoincrement, do a simple sort.
1094 assert not composite_sentinel
1095 result.extend(
1096 sorted(rows, key=operator.itemgetter(-1))
1097 )
1098 continue
1100 # otherwise, create dictionaries to match up batches
1101 # with parameters
1102 assert imv.sentinel_param_keys
1103 assert imv.sentinel_columns
1105 _nsc = imv.num_sentinel_columns
1107 if not _sentinel_proc_initialized:
1108 if composite_sentinel:
1109 _composite_sentinel_proc = [
1110 col.type._cached_result_processor(
1111 self, cursor_desc[1]
1112 )
1113 for col, cursor_desc in zip(
1114 imv.sentinel_columns,
1115 cursor.description[-_nsc:],
1116 )
1117 ]
1118 else:
1119 _scalar_sentinel_proc = (
1120 imv.sentinel_columns[0]
1121 ).type._cached_result_processor(
1122 self, cursor.description[-1][1]
1123 )
1124 _sentinel_proc_initialized = True
1126 rows_by_sentinel: Union[
1127 Dict[Tuple[Any, ...], Any],
1128 Dict[Any, Any],
1129 ]
1131 if composite_sentinel:
1132 rows_by_sentinel = {
1133 tuple(
1134 (proc(val) if proc else val)
1135 for val, proc in zip(
1136 row[-_nsc:], _composite_sentinel_proc
1137 )
1138 ): row
1139 for row in rows
1140 }
1141 elif _scalar_sentinel_proc:
1142 rows_by_sentinel = {
1143 _scalar_sentinel_proc(row[-1]): row for row in rows
1144 }
1145 else:
1146 rows_by_sentinel = {row[-1]: row for row in rows}
1148 if len(rows_by_sentinel) != len(imv_batch.batch):
1149 # see test_insert_exec.py::
1150 # IMVSentinelTest::test_sentinel_incorrect_rowcount
1151 # for coverage / demonstration
1152 raise exc.InvalidRequestError(
1153 f"Sentinel-keyed result set did not produce "
1154 f"correct number of rows {len(imv_batch.batch)}; "
1155 "produced "
1156 f"{len(rows_by_sentinel)}. Please ensure the "
1157 "sentinel column is fully unique and populated in "
1158 "all cases."
1159 )
1161 try:
1162 ordered_rows = [
1163 rows_by_sentinel[sentinel_keys]
1164 for sentinel_keys in imv_batch.sentinel_values
1165 ]
1166 except KeyError as ke:
1167 # see test_insert_exec.py::
1168 # IMVSentinelTest::test_sentinel_cant_match_keys
1169 # for coverage / demonstration
1170 raise exc.InvalidRequestError(
1171 f"Can't match sentinel values in result set to "
1172 f"parameter sets; key {ke.args[0]!r} was not "
1173 "found. "
1174 "There may be a mismatch between the datatype "
1175 "passed to the DBAPI driver vs. that which it "
1176 "returns in a result row. Ensure the given "
1177 "Python value matches the expected result type "
1178 "*exactly*, taking care to not rely upon implicit "
1179 "conversions which may occur such as when using "
1180 "strings in place of UUID or integer values, etc. "
1181 ) from ke
1183 result.extend(ordered_rows)
1185 else:
1186 result.extend(rows)
1188 def do_executemany(self, cursor, statement, parameters, context=None):
1189 cursor.executemany(statement, parameters)
1191 def do_execute(self, cursor, statement, parameters, context=None):
1192 cursor.execute(statement, parameters)
1194 def do_execute_no_params(self, cursor, statement, context=None):
1195 cursor.execute(statement)
1197 def is_disconnect(
1198 self,
1199 e: DBAPIModule.Error,
1200 connection: Union[
1201 pool.PoolProxiedConnection, interfaces.DBAPIConnection, None
1202 ],
1203 cursor: Optional[interfaces.DBAPICursor],
1204 ) -> bool:
1205 return False
1207 @util.memoized_instancemethod
1208 def _gen_allowed_isolation_levels(self, dbapi_conn):
1209 try:
1210 raw_levels = list(self.get_isolation_level_values(dbapi_conn))
1211 except NotImplementedError:
1212 return None
1213 else:
1214 normalized_levels = [
1215 level.replace("_", " ").upper() for level in raw_levels
1216 ]
1217 if raw_levels != normalized_levels:
1218 raise ValueError(
1219 f"Dialect {self.name!r} get_isolation_level_values() "
1220 f"method should return names as UPPERCASE using spaces, "
1221 f"not underscores; got "
1222 f"{sorted(set(raw_levels).difference(normalized_levels))}"
1223 )
1224 return tuple(normalized_levels)
1226 def _assert_and_set_isolation_level(self, dbapi_conn, level):
1227 level = level.replace("_", " ").upper()
1229 _allowed_isolation_levels = self._gen_allowed_isolation_levels(
1230 dbapi_conn
1231 )
1232 if (
1233 _allowed_isolation_levels
1234 and level not in _allowed_isolation_levels
1235 ):
1236 raise exc.ArgumentError(
1237 f"Invalid value {level!r} for isolation_level. "
1238 f"Valid isolation levels for {self.name!r} are "
1239 f"{', '.join(_allowed_isolation_levels)}"
1240 )
1242 self.set_isolation_level(dbapi_conn, level)
1244 def reset_isolation_level(self, dbapi_conn):
1245 if self._on_connect_isolation_level is not None:
1246 assert (
1247 self._on_connect_isolation_level == "AUTOCOMMIT"
1248 or self._on_connect_isolation_level
1249 == self.default_isolation_level
1250 )
1251 self._assert_and_set_isolation_level(
1252 dbapi_conn, self._on_connect_isolation_level
1253 )
1254 else:
1255 assert self.default_isolation_level is not None
1256 self._assert_and_set_isolation_level(
1257 dbapi_conn,
1258 self.default_isolation_level,
1259 )
1261 def normalize_name(self, name):
1262 if name is None:
1263 return None
1265 name_lower = name.lower()
1266 name_upper = name.upper()
1268 if name_upper == name_lower:
1269 # name has no upper/lower conversion, e.g. non-european characters.
1270 # return unchanged
1271 return name
1272 elif name_upper == name and not (
1273 self.identifier_preparer._requires_quotes
1274 )(name_lower):
1275 # name is all uppercase and doesn't require quoting; normalize
1276 # to all lower case
1277 return name_lower
1278 elif name_lower == name:
1279 # name is all lower case, which if denormalized means we need to
1280 # force quoting on it
1281 return quoted_name(name, quote=True)
1282 else:
1283 # name is mixed case, means it will be quoted in SQL when used
1284 # later, no normalizes
1285 return name
1287 def denormalize_name(self, name):
1288 if name is None:
1289 return None
1291 name_lower = name.lower()
1292 name_upper = name.upper()
1294 if name_upper == name_lower:
1295 # name has no upper/lower conversion, e.g. non-european characters.
1296 # return unchanged
1297 return name
1298 elif name_lower == name and not (
1299 self.identifier_preparer._requires_quotes
1300 )(name_lower):
1301 name = name_upper
1302 return name
1304 def get_driver_connection(self, connection: DBAPIConnection) -> Any:
1305 return connection
1307 def _overrides_default(self, method):
1308 return (
1309 getattr(type(self), method).__code__
1310 is not getattr(DefaultDialect, method).__code__
1311 )
1313 def _default_multi_reflect(
1314 self,
1315 single_tbl_method,
1316 connection,
1317 kind,
1318 schema,
1319 filter_names,
1320 scope,
1321 **kw,
1322 ):
1323 names_fns = []
1324 temp_names_fns = []
1325 if ObjectKind.TABLE in kind:
1326 names_fns.append(self.get_table_names)
1327 temp_names_fns.append(self.get_temp_table_names)
1328 if ObjectKind.VIEW in kind:
1329 names_fns.append(self.get_view_names)
1330 temp_names_fns.append(self.get_temp_view_names)
1331 if ObjectKind.MATERIALIZED_VIEW in kind:
1332 names_fns.append(self.get_materialized_view_names)
1333 # no temp materialized view at the moment
1334 # temp_names_fns.append(self.get_temp_materialized_view_names)
1336 unreflectable = kw.pop("unreflectable", {})
1338 if (
1339 filter_names
1340 and scope is ObjectScope.ANY
1341 and kind is ObjectKind.ANY
1342 ):
1343 # if names are given and no qualification on type of table
1344 # (i.e. the Table(..., autoload) case), take the names as given,
1345 # don't run names queries. If a table does not exit
1346 # NoSuchTableError is raised and it's skipped
1348 # this also suits the case for mssql where we can reflect
1349 # individual temp tables but there's no temp_names_fn
1350 names = filter_names
1351 else:
1352 names = []
1353 name_kw = {"schema": schema, **kw}
1354 fns = []
1355 if ObjectScope.DEFAULT in scope:
1356 fns.extend(names_fns)
1357 if ObjectScope.TEMPORARY in scope:
1358 fns.extend(temp_names_fns)
1360 for fn in fns:
1361 try:
1362 names.extend(fn(connection, **name_kw))
1363 except NotImplementedError:
1364 pass
1366 if filter_names:
1367 filter_names = set(filter_names)
1369 # iterate over all the tables/views and call the single table method
1370 for table in names:
1371 if not filter_names or table in filter_names:
1372 key = (schema, table)
1373 try:
1374 yield (
1375 key,
1376 single_tbl_method(
1377 connection, table, schema=schema, **kw
1378 ),
1379 )
1380 except exc.UnreflectableTableError as err:
1381 if key not in unreflectable:
1382 unreflectable[key] = err
1383 except exc.NoSuchTableError:
1384 pass
1386 def has_multi_table(
1387 self,
1388 connection: Connection,
1389 table_names: Sequence[str],
1390 schema: Optional[str] = None,
1391 **kw: Any,
1392 ) -> Iterable[tuple[TableKey, bool]]:
1393 for table_name in table_names:
1394 exist = self.has_table(connection, table_name, schema=schema, **kw)
1395 yield (schema, table_name), exist
1397 def get_multi_table_options(self, connection, **kw):
1398 return self._default_multi_reflect(
1399 self.get_table_options, connection, **kw
1400 )
1402 def get_multi_columns(self, connection, **kw):
1403 return self._default_multi_reflect(self.get_columns, connection, **kw)
1405 def get_multi_pk_constraint(self, connection, **kw):
1406 return self._default_multi_reflect(
1407 self.get_pk_constraint, connection, **kw
1408 )
1410 def get_multi_foreign_keys(self, connection, **kw):
1411 return self._default_multi_reflect(
1412 self.get_foreign_keys, connection, **kw
1413 )
1415 def get_multi_indexes(self, connection, **kw):
1416 return self._default_multi_reflect(self.get_indexes, connection, **kw)
1418 def get_multi_unique_constraints(self, connection, **kw):
1419 return self._default_multi_reflect(
1420 self.get_unique_constraints, connection, **kw
1421 )
1423 def get_multi_check_constraints(self, connection, **kw):
1424 return self._default_multi_reflect(
1425 self.get_check_constraints, connection, **kw
1426 )
1428 def get_multi_table_comment(self, connection, **kw):
1429 return self._default_multi_reflect(
1430 self.get_table_comment, connection, **kw
1431 )
1434class StrCompileDialect(DefaultDialect):
1435 statement_compiler = compiler.StrSQLCompiler
1436 ddl_compiler = compiler.DDLCompiler
1437 type_compiler_cls = compiler.StrSQLTypeCompiler
1438 preparer = compiler.IdentifierPreparer
1440 insert_returning = True
1441 update_returning = True
1442 delete_returning = True
1444 supports_statement_cache = True
1446 supports_identity_columns = True
1448 supports_sequences = True
1449 sequences_optional = True
1450 preexecute_autoincrement_sequences = False
1452 supports_native_boolean = True
1454 supports_multivalues_insert = True
1455 supports_simple_order_by_label = True
1458class DefaultExecutionContext(ExecutionContext):
1459 isinsert = False
1460 isupdate = False
1461 isdelete = False
1462 is_crud = False
1463 is_text = False
1464 isddl = False
1466 execute_style: ExecuteStyle = ExecuteStyle.EXECUTE
1468 compiled: Optional[Compiled] = None
1469 result_column_struct: Optional[
1470 Tuple[List[ResultColumnsEntry], bool, bool, bool, bool]
1471 ] = None
1472 returned_default_rows: Optional[Sequence[Row[Unpack[TupleAny]]]] = None
1474 execution_options: _ExecuteOptions = util.EMPTY_DICT
1476 cursor_fetch_strategy = _cursor._DEFAULT_FETCH
1478 invoked_statement: Optional[Executable] = None
1480 _is_implicit_returning = False
1481 _is_explicit_returning = False
1482 _is_supplemental_returning = False
1483 _is_server_side = False
1485 _soft_closed = False
1487 _rowcount: Optional[int] = None
1489 # a hook for SQLite's translation of
1490 # result column names
1491 # NOTE: pyhive is using this hook, can't remove it :(
1492 _translate_colname: Optional[
1493 Callable[[str], Tuple[str, Optional[str]]]
1494 ] = None
1496 _expanded_parameters: Mapping[str, List[str]] = util.immutabledict()
1497 """used by set_input_sizes().
1499 This collection comes from ``ExpandedState.parameter_expansion``.
1501 """
1503 cache_hit = NO_CACHE_KEY
1505 root_connection: Connection
1506 _dbapi_connection: PoolProxiedConnection
1507 dialect: Dialect
1508 unicode_statement: str
1509 cursor: DBAPICursor
1510 compiled_parameters: List[_MutableCoreSingleExecuteParams]
1511 parameters: _DBAPIMultiExecuteParams
1512 extracted_parameters: Optional[Sequence[BindParameter[Any]]]
1514 _empty_dict_params = cast("Mapping[str, Any]", util.EMPTY_DICT)
1516 _insertmanyvalues_rows: Optional[List[Tuple[Any, ...]]] = None
1517 _num_sentinel_cols: int = 0
1519 @classmethod
1520 def _init_ddl(
1521 cls,
1522 dialect: Dialect,
1523 connection: Connection,
1524 dbapi_connection: PoolProxiedConnection,
1525 execution_options: _ExecuteOptions,
1526 compiled_ddl: DDLCompiler,
1527 ) -> ExecutionContext:
1528 """Initialize execution context for an ExecutableDDLElement
1529 construct."""
1531 self = cls.__new__(cls)
1532 self.root_connection = connection
1533 self._dbapi_connection = dbapi_connection
1534 self.dialect = connection.dialect
1536 self.compiled = compiled = compiled_ddl
1537 self.isddl = True
1539 self.execution_options = execution_options
1541 self.unicode_statement = str(compiled)
1542 if compiled.schema_translate_map:
1543 schema_translate_map = self.execution_options.get(
1544 "schema_translate_map", {}
1545 )
1547 rst = compiled.preparer._render_schema_translates
1548 self.unicode_statement = rst(
1549 self.unicode_statement, schema_translate_map
1550 )
1552 self.statement = self.unicode_statement
1554 self.cursor = self.create_cursor()
1555 self.compiled_parameters = []
1557 if dialect.positional:
1558 self.parameters = [dialect.execute_sequence_format()]
1559 else:
1560 self.parameters = [self._empty_dict_params]
1562 return self
1564 @classmethod
1565 def _init_compiled(
1566 cls,
1567 dialect: Dialect,
1568 connection: Connection,
1569 dbapi_connection: PoolProxiedConnection,
1570 execution_options: _ExecuteOptions,
1571 compiled: SQLCompiler,
1572 parameters: _CoreMultiExecuteParams,
1573 invoked_statement: Executable,
1574 extracted_parameters: Optional[Sequence[BindParameter[Any]]],
1575 cache_hit: CacheStats = CacheStats.CACHING_DISABLED,
1576 param_dict: _CoreSingleExecuteParams | None = None,
1577 ) -> ExecutionContext:
1578 """Initialize execution context for a Compiled construct."""
1580 self = cls.__new__(cls)
1581 self.root_connection = connection
1582 self._dbapi_connection = dbapi_connection
1583 self.dialect = connection.dialect
1584 self.extracted_parameters = extracted_parameters
1585 self.invoked_statement = invoked_statement
1586 self.compiled = compiled
1587 self.cache_hit = cache_hit
1589 self.execution_options = execution_options
1591 self.result_column_struct = (
1592 compiled._result_columns,
1593 compiled._ordered_columns,
1594 compiled._textual_ordered_columns,
1595 compiled._ad_hoc_textual,
1596 compiled._loose_column_name_matching,
1597 )
1599 self.isinsert = ii = compiled.isinsert
1600 self.isupdate = iu = compiled.isupdate
1601 self.isdelete = id_ = compiled.isdelete
1602 self.is_text = compiled.isplaintext
1604 if ii or iu or id_:
1605 dml_statement = compiled.compile_state.statement # type: ignore[union-attr] # noqa: E501
1606 if TYPE_CHECKING:
1607 assert isinstance(dml_statement, UpdateBase)
1608 self.is_crud = True
1609 self._is_explicit_returning = ier = bool(dml_statement._returning)
1610 self._is_implicit_returning = iir = bool(
1611 compiled.implicit_returning
1612 )
1613 if iir and dml_statement._supplemental_returning:
1614 self._is_supplemental_returning = True
1616 # dont mix implicit and explicit returning
1617 assert not (iir and ier)
1619 if (ier or iir) and compiled.for_executemany:
1620 if ii and not self.dialect.insert_executemany_returning:
1621 raise exc.InvalidRequestError(
1622 f"Dialect {self.dialect.dialect_description} with "
1623 f"current server capabilities does not support "
1624 "INSERT..RETURNING when executemany is used"
1625 )
1626 elif (
1627 ii
1628 and dml_statement._sort_by_parameter_order
1629 and not self.dialect.insert_executemany_returning_sort_by_parameter_order # noqa: E501
1630 ):
1631 raise exc.InvalidRequestError(
1632 f"Dialect {self.dialect.dialect_description} with "
1633 f"current server capabilities does not support "
1634 "INSERT..RETURNING with deterministic row ordering "
1635 "when executemany is used"
1636 )
1637 elif (
1638 ii
1639 and self.dialect.use_insertmanyvalues
1640 and not compiled._insertmanyvalues
1641 ):
1642 raise exc.InvalidRequestError(
1643 'Statement does not have "insertmanyvalues" '
1644 "enabled, can't use INSERT..RETURNING with "
1645 "executemany in this case."
1646 )
1647 elif iu and not self.dialect.update_executemany_returning:
1648 raise exc.InvalidRequestError(
1649 f"Dialect {self.dialect.dialect_description} with "
1650 f"current server capabilities does not support "
1651 "UPDATE..RETURNING when executemany is used"
1652 )
1653 elif id_ and not self.dialect.delete_executemany_returning:
1654 raise exc.InvalidRequestError(
1655 f"Dialect {self.dialect.dialect_description} with "
1656 f"current server capabilities does not support "
1657 "DELETE..RETURNING when executemany is used"
1658 )
1660 if not parameters:
1661 self.compiled_parameters = [
1662 compiled.construct_params(
1663 extracted_parameters=extracted_parameters,
1664 escape_names=False,
1665 _collected_params=param_dict,
1666 )
1667 ]
1668 else:
1669 self.compiled_parameters = [
1670 compiled.construct_params(
1671 m,
1672 escape_names=False,
1673 _group_number=grp,
1674 extracted_parameters=extracted_parameters,
1675 _collected_params=param_dict,
1676 )
1677 for grp, m in enumerate(parameters)
1678 ]
1680 if len(parameters) > 1:
1681 if self.isinsert and compiled._insertmanyvalues:
1682 self.execute_style = ExecuteStyle.INSERTMANYVALUES
1684 imv = compiled._insertmanyvalues
1685 if imv.sentinel_columns is not None:
1686 self._num_sentinel_cols = imv.num_sentinel_columns
1687 else:
1688 self.execute_style = ExecuteStyle.EXECUTEMANY
1690 self.unicode_statement = compiled.string
1692 self.cursor = self.create_cursor()
1694 if self.compiled.insert_prefetch or self.compiled.update_prefetch:
1695 self._process_execute_defaults()
1697 processors = compiled._bind_processors
1699 flattened_processors: Mapping[
1700 str, _BindProcessorType[Any]
1701 ] = processors # type: ignore[assignment]
1703 if compiled.literal_execute_params or compiled.post_compile_params:
1704 if self.executemany:
1705 raise exc.InvalidRequestError(
1706 "'literal_execute' or 'expanding' parameters can't be "
1707 "used with executemany()"
1708 )
1710 expanded_state = compiled._process_parameters_for_postcompile(
1711 self.compiled_parameters[0]
1712 )
1714 # re-assign self.unicode_statement
1715 self.unicode_statement = expanded_state.statement
1717 self._expanded_parameters = expanded_state.parameter_expansion
1719 flattened_processors = dict(processors) # type: ignore[arg-type]
1720 flattened_processors.update(expanded_state.processors)
1721 positiontup = expanded_state.positiontup
1722 elif compiled.positional:
1723 positiontup = self.compiled.positiontup
1724 else:
1725 positiontup = None
1727 if compiled.schema_translate_map:
1728 schema_translate_map = self.execution_options.get(
1729 "schema_translate_map", {}
1730 )
1731 rst = compiled.preparer._render_schema_translates
1732 self.unicode_statement = rst(
1733 self.unicode_statement, schema_translate_map
1734 )
1736 # final self.unicode_statement is now assigned, encode if needed
1737 # by dialect
1738 self.statement = self.unicode_statement
1740 # Convert the dictionary of bind parameter values
1741 # into a dict or list to be sent to the DBAPI's
1742 # execute() or executemany() method.
1744 if compiled.positional:
1745 core_positional_parameters: MutableSequence[Sequence[Any]] = []
1746 assert positiontup is not None
1747 for compiled_params in self.compiled_parameters:
1748 l_param: List[Any] = [
1749 (
1750 flattened_processors[key](compiled_params[key])
1751 if key in flattened_processors
1752 else compiled_params[key]
1753 )
1754 for key in positiontup
1755 ]
1756 core_positional_parameters.append(
1757 dialect.execute_sequence_format(l_param)
1758 )
1760 self.parameters = core_positional_parameters
1761 else:
1762 core_dict_parameters: MutableSequence[Dict[str, Any]] = []
1763 escaped_names = compiled.escaped_bind_names
1765 # note that currently, "expanded" parameters will be present
1766 # in self.compiled_parameters in their quoted form. This is
1767 # slightly inconsistent with the approach taken as of
1768 # #8056 where self.compiled_parameters is meant to contain unquoted
1769 # param names.
1770 d_param: Dict[str, Any]
1771 for compiled_params in self.compiled_parameters:
1772 if escaped_names:
1773 d_param = {
1774 escaped_names.get(key, key): (
1775 flattened_processors[key](compiled_params[key])
1776 if key in flattened_processors
1777 else compiled_params[key]
1778 )
1779 for key in compiled_params
1780 }
1781 else:
1782 d_param = {
1783 key: (
1784 flattened_processors[key](compiled_params[key])
1785 if key in flattened_processors
1786 else compiled_params[key]
1787 )
1788 for key in compiled_params
1789 }
1791 core_dict_parameters.append(d_param)
1793 self.parameters = core_dict_parameters
1795 return self
1797 @classmethod
1798 def _init_statement(
1799 cls,
1800 dialect: Dialect,
1801 connection: Connection,
1802 dbapi_connection: PoolProxiedConnection,
1803 execution_options: _ExecuteOptions,
1804 statement: str,
1805 parameters: _DBAPIMultiExecuteParams,
1806 ) -> ExecutionContext:
1807 """Initialize execution context for a string SQL statement."""
1809 self = cls.__new__(cls)
1810 self.root_connection = connection
1811 self._dbapi_connection = dbapi_connection
1812 self.dialect = connection.dialect
1813 self.is_text = True
1815 self.execution_options = execution_options
1817 if not parameters:
1818 if self.dialect.positional:
1819 self.parameters = [dialect.execute_sequence_format()]
1820 else:
1821 self.parameters = [self._empty_dict_params]
1822 elif isinstance(parameters[0], dialect.execute_sequence_format):
1823 self.parameters = parameters
1824 elif isinstance(parameters[0], dict):
1825 self.parameters = parameters
1826 else:
1827 self.parameters = [
1828 dialect.execute_sequence_format(p) for p in parameters
1829 ]
1831 if len(parameters) > 1:
1832 self.execute_style = ExecuteStyle.EXECUTEMANY
1834 self.statement = self.unicode_statement = statement
1836 self.cursor = self.create_cursor()
1837 return self
1839 @classmethod
1840 def _init_default(
1841 cls,
1842 dialect: Dialect,
1843 connection: Connection,
1844 dbapi_connection: PoolProxiedConnection,
1845 execution_options: _ExecuteOptions,
1846 ) -> ExecutionContext:
1847 """Initialize execution context for a ColumnDefault construct."""
1849 self = cls.__new__(cls)
1850 self.root_connection = connection
1851 self._dbapi_connection = dbapi_connection
1852 self.dialect = connection.dialect
1854 self.execution_options = execution_options
1856 self.cursor = self.create_cursor()
1857 return self
1859 def _get_cache_stats(self) -> str:
1860 if self.compiled is None:
1861 return "raw sql"
1863 now = perf_counter()
1865 ch = self.cache_hit
1867 gen_time = self.compiled._gen_time
1868 assert gen_time is not None
1870 if ch is NO_CACHE_KEY:
1871 return "no key %.5fs" % (now - gen_time,)
1872 elif ch is CACHE_HIT:
1873 return "cached since %.4gs ago" % (now - gen_time,)
1874 elif ch is CACHE_MISS:
1875 return "generated in %.5fs" % (now - gen_time,)
1876 elif ch is CACHING_DISABLED:
1877 if "_cache_disable_reason" in self.execution_options:
1878 return "caching disabled (%s) %.5fs " % (
1879 self.execution_options["_cache_disable_reason"],
1880 now - gen_time,
1881 )
1882 else:
1883 return "caching disabled %.5fs" % (now - gen_time,)
1884 elif ch is NO_DIALECT_SUPPORT:
1885 return "dialect %s+%s does not support caching %.5fs" % (
1886 self.dialect.name,
1887 self.dialect.driver,
1888 now - gen_time,
1889 )
1890 else:
1891 return "unknown"
1893 @property
1894 def executemany(self): # type: ignore[override]
1895 return self.execute_style in (
1896 ExecuteStyle.EXECUTEMANY,
1897 ExecuteStyle.INSERTMANYVALUES,
1898 )
1900 @util.memoized_property
1901 def identifier_preparer(self):
1902 if self.compiled:
1903 return self.compiled.preparer
1904 elif "schema_translate_map" in self.execution_options:
1905 return self.dialect.identifier_preparer._with_schema_translate(
1906 self.execution_options["schema_translate_map"]
1907 )
1908 else:
1909 return self.dialect.identifier_preparer
1911 @util.memoized_property
1912 def engine(self):
1913 return self.root_connection.engine
1915 @util.memoized_property
1916 def postfetch_cols(self) -> Optional[Sequence[Column[Any]]]:
1917 if TYPE_CHECKING:
1918 assert isinstance(self.compiled, SQLCompiler)
1919 return self.compiled.postfetch
1921 @util.memoized_property
1922 def prefetch_cols(self) -> Optional[Sequence[Column[Any]]]:
1923 if TYPE_CHECKING:
1924 assert isinstance(self.compiled, SQLCompiler)
1925 if self.isinsert:
1926 return self.compiled.insert_prefetch
1927 elif self.isupdate:
1928 return self.compiled.update_prefetch
1929 else:
1930 return ()
1932 @util.memoized_property
1933 def no_parameters(self):
1934 return self.execution_options.get("no_parameters", False)
1936 def _execute_scalar(
1937 self,
1938 stmt: str,
1939 type_: Optional[TypeEngine[Any]],
1940 parameters: Optional[_DBAPISingleExecuteParams] = None,
1941 ) -> Any:
1942 """Execute a string statement on the current cursor, returning a
1943 scalar result.
1945 Used to fire off sequences, default phrases, and "select lastrowid"
1946 types of statements individually or in the context of a parent INSERT
1947 or UPDATE statement.
1949 """
1951 conn = self.root_connection
1953 if "schema_translate_map" in self.execution_options:
1954 schema_translate_map = self.execution_options.get(
1955 "schema_translate_map", {}
1956 )
1958 rst = self.identifier_preparer._render_schema_translates
1959 stmt = rst(stmt, schema_translate_map)
1961 if not parameters:
1962 if self.dialect.positional:
1963 parameters = self.dialect.execute_sequence_format()
1964 else:
1965 parameters = {}
1967 conn._cursor_execute(self.cursor, stmt, parameters, context=self)
1968 row = self.cursor.fetchone()
1969 if row is not None:
1970 r = row[0]
1971 else:
1972 r = None
1973 if type_ is not None:
1974 # apply type post processors to the result
1975 proc = type_._cached_result_processor(
1976 self.dialect, self.cursor.description[0][1]
1977 )
1978 if proc:
1979 return proc(r)
1980 return r
1982 @util.memoized_property
1983 def connection(self):
1984 return self.root_connection
1986 def _use_server_side_cursor(self):
1987 if not self.dialect.supports_server_side_cursors:
1988 return False
1990 if self.dialect.server_side_cursors:
1991 # this is deprecated
1992 use_server_side = self.execution_options.get(
1993 "stream_results", True
1994 ) and (
1995 self.compiled
1996 and isinstance(self.compiled.statement, expression.Selectable)
1997 or (
1998 (
1999 not self.compiled
2000 or isinstance(
2001 self.compiled.statement, expression.TextClause
2002 )
2003 )
2004 and self.unicode_statement
2005 and SERVER_SIDE_CURSOR_RE.match(self.unicode_statement)
2006 )
2007 )
2008 else:
2009 use_server_side = self.execution_options.get(
2010 "stream_results", False
2011 )
2013 return use_server_side
2015 def create_cursor(self) -> DBAPICursor:
2016 if (
2017 # inlining initial preference checks for SS cursors
2018 self.dialect.supports_server_side_cursors
2019 and (
2020 self.execution_options.get("stream_results", False)
2021 or (
2022 self.dialect.server_side_cursors
2023 and self._use_server_side_cursor()
2024 )
2025 )
2026 ):
2027 self._is_server_side = True
2028 return self.create_server_side_cursor()
2029 else:
2030 self._is_server_side = False
2031 return self.create_default_cursor()
2033 def fetchall_for_returning(self, cursor):
2034 return cursor.fetchall()
2036 def create_default_cursor(self) -> DBAPICursor:
2037 return self._dbapi_connection.cursor()
2039 def create_server_side_cursor(self) -> DBAPICursor:
2040 raise NotImplementedError()
2042 def pre_exec(self):
2043 pass
2045 def get_out_parameter_values(self, names):
2046 raise NotImplementedError(
2047 "This dialect does not support OUT parameters"
2048 )
2050 def post_exec(self):
2051 pass
2053 def get_result_processor(
2054 self, type_: TypeEngine[Any], colname: str, coltype: DBAPIType
2055 ) -> Optional[_ResultProcessorType[Any]]:
2056 """Return a 'result processor' for a given type as present in
2057 cursor.description.
2059 This has a default implementation that dialects can override
2060 for context-sensitive result type handling.
2062 """
2063 return type_._cached_result_processor(self.dialect, coltype)
2065 def get_lastrowid(self) -> int:
2066 """return self.cursor.lastrowid, or equivalent, after an INSERT.
2068 This may involve calling special cursor functions, issuing a new SELECT
2069 on the cursor (or a new one), or returning a stored value that was
2070 calculated within post_exec().
2072 This function will only be called for dialects which support "implicit"
2073 primary key generation, keep preexecute_autoincrement_sequences set to
2074 False, and when no explicit id value was bound to the statement.
2076 The function is called once for an INSERT statement that would need to
2077 return the last inserted primary key for those dialects that make use
2078 of the lastrowid concept. In these cases, it is called directly after
2079 :meth:`.ExecutionContext.post_exec`.
2081 """
2082 return self.cursor.lastrowid
2084 def handle_dbapi_exception(self, e):
2085 pass
2087 @util.non_memoized_property
2088 def rowcount(self) -> int:
2089 if self._rowcount is not None:
2090 return self._rowcount
2091 else:
2092 return self.cursor.rowcount
2094 @property
2095 def _has_rowcount(self):
2096 return self._rowcount is not None
2098 def supports_sane_rowcount(self):
2099 return self.dialect.supports_sane_rowcount
2101 def supports_sane_multi_rowcount(self):
2102 return self.dialect.supports_sane_multi_rowcount
2104 def _setup_result_proxy(self):
2105 exec_opt = self.execution_options
2107 if self._rowcount is None and exec_opt.get("preserve_rowcount", False):
2108 self._rowcount = self.cursor.rowcount
2110 yp: Optional[Union[int, bool]]
2111 if self.is_crud or self.is_text:
2112 result = self._setup_dml_or_text_result()
2113 yp = False
2114 else:
2115 yp = exec_opt.get("yield_per", None)
2116 sr = self._is_server_side or exec_opt.get("stream_results", False)
2117 strategy = self.cursor_fetch_strategy
2118 if sr and strategy is _cursor._DEFAULT_FETCH:
2119 strategy = _cursor.BufferedRowCursorFetchStrategy(
2120 self.cursor, self.execution_options
2121 )
2122 cursor_description: _DBAPICursorDescription = (
2123 strategy.alternate_cursor_description
2124 or self.cursor.description
2125 )
2126 if cursor_description is None:
2127 strategy = _cursor._NO_CURSOR_DQL
2129 result = _cursor.CursorResult(self, strategy, cursor_description)
2131 compiled = self.compiled
2133 if (
2134 compiled
2135 and not self.isddl
2136 and cast(SQLCompiler, compiled).has_out_parameters
2137 ):
2138 self._setup_out_parameters(result)
2140 if not result._soft_closed and not result._metadata.returns_rows:
2141 # no results, get rowcount (which requires open cursor on
2142 # some drivers) and release the cursor. this applies to all
2143 # statements that don't return rows, including DML, text(),
2144 # DDL and SAVEPOINT
2145 if self._rowcount is None:
2146 self._rowcount = self.cursor.rowcount
2147 result._soft_close()
2149 self._soft_closed = result._soft_closed
2151 if yp:
2152 result = result.yield_per(yp)
2154 return result
2156 def _setup_out_parameters(self, result):
2157 compiled = cast(SQLCompiler, self.compiled)
2159 out_bindparams = [
2160 (param, name)
2161 for param, name in compiled.bind_names.items()
2162 if param.isoutparam
2163 ]
2164 out_parameters = {}
2166 for bindparam, raw_value in zip(
2167 [param for param, name in out_bindparams],
2168 self.get_out_parameter_values(
2169 [name for param, name in out_bindparams]
2170 ),
2171 ):
2172 type_ = bindparam.type
2173 impl_type = type_.dialect_impl(self.dialect)
2174 dbapi_type = impl_type.get_dbapi_type(self.dialect.loaded_dbapi)
2175 result_processor = impl_type.result_processor(
2176 self.dialect, dbapi_type
2177 )
2178 if result_processor is not None:
2179 raw_value = result_processor(raw_value)
2180 out_parameters[bindparam.key] = raw_value
2182 result.out_parameters = out_parameters
2184 def _setup_dml_or_text_result(self):
2185 compiled = cast(SQLCompiler, self.compiled)
2187 strategy: ResultFetchStrategy = self.cursor_fetch_strategy
2189 if self.isinsert:
2190 if (
2191 self.execute_style is ExecuteStyle.INSERTMANYVALUES
2192 and compiled.effective_returning
2193 ):
2194 strategy = _cursor.FullyBufferedCursorFetchStrategy(
2195 self.cursor,
2196 initial_buffer=self._insertmanyvalues_rows,
2197 # maintain alt cursor description if set by the
2198 # dialect, e.g. mssql preserves it
2199 alternate_description=(
2200 strategy.alternate_cursor_description
2201 ),
2202 )
2204 if compiled.postfetch_lastrowid:
2205 self.inserted_primary_key_rows = (
2206 self._setup_ins_pk_from_lastrowid()
2207 )
2208 # else if not self._is_implicit_returning,
2209 # the default inserted_primary_key_rows accessor will
2210 # return an "empty" primary key collection when accessed.
2212 if self._is_server_side and strategy is _cursor._DEFAULT_FETCH:
2213 strategy = _cursor.BufferedRowCursorFetchStrategy(
2214 self.cursor, self.execution_options
2215 )
2217 if strategy is _cursor._NO_CURSOR_DML:
2218 cursor_description = None
2219 else:
2220 cursor_description = (
2221 strategy.alternate_cursor_description
2222 or self.cursor.description
2223 )
2225 if cursor_description is None:
2226 strategy = _cursor._NO_CURSOR_DML
2227 elif self._num_sentinel_cols:
2228 assert self.execute_style is ExecuteStyle.INSERTMANYVALUES
2229 # the sentinel columns are handled in CursorResult._init_metadata
2230 # using essentially _reduce
2232 result: _cursor.CursorResult[Any] = _cursor.CursorResult(
2233 self, strategy, cursor_description
2234 )
2236 if self.isinsert:
2237 if self._is_implicit_returning:
2238 rows = result.all()
2240 self.returned_default_rows = rows
2242 self.inserted_primary_key_rows = (
2243 self._setup_ins_pk_from_implicit_returning(result, rows)
2244 )
2246 # test that it has a cursor metadata that is accurate. the
2247 # first row will have been fetched and current assumptions
2248 # are that the result has only one row, until executemany()
2249 # support is added here.
2250 assert result._metadata.returns_rows
2252 # Insert statement has both return_defaults() and
2253 # returning(). rewind the result on the list of rows
2254 # we just used.
2255 if self._is_supplemental_returning:
2256 result._rewind(rows)
2257 else:
2258 result._soft_close()
2259 elif not self._is_explicit_returning:
2260 result._soft_close()
2262 # we assume here the result does not return any rows.
2263 # *usually*, this will be true. However, some dialects
2264 # such as that of MSSQL/pyodbc need to SELECT a post fetch
2265 # function so this is not necessarily true.
2266 # assert not result.returns_rows
2268 elif self._is_implicit_returning:
2269 rows = result.all()
2271 if rows:
2272 self.returned_default_rows = rows
2273 self._rowcount = len(rows)
2275 if self._is_supplemental_returning:
2276 result._rewind(rows)
2277 else:
2278 result._soft_close()
2280 # test that it has a cursor metadata that is accurate.
2281 # the rows have all been fetched however.
2282 assert result._metadata.returns_rows
2284 elif self.isupdate or self.isdelete:
2285 if self._rowcount is None:
2286 self._rowcount = self.cursor.rowcount
2287 return result
2289 @util.memoized_property
2290 def inserted_primary_key_rows(self):
2291 # if no specific "get primary key" strategy was set up
2292 # during execution, return a "default" primary key based
2293 # on what's in the compiled_parameters and nothing else.
2294 return self._setup_ins_pk_from_empty()
2296 def _setup_ins_pk_from_lastrowid(self):
2297 getter = cast(
2298 SQLCompiler, self.compiled
2299 )._inserted_primary_key_from_lastrowid_getter
2300 lastrowid = self.get_lastrowid()
2301 return [getter(lastrowid, self.compiled_parameters[0])]
2303 def _setup_ins_pk_from_empty(self):
2304 getter = cast(
2305 SQLCompiler, self.compiled
2306 )._inserted_primary_key_from_lastrowid_getter
2307 return [getter(None, param) for param in self.compiled_parameters]
2309 def _setup_ins_pk_from_implicit_returning(self, result, rows):
2310 if not rows:
2311 return []
2313 getter = cast(
2314 SQLCompiler, self.compiled
2315 )._inserted_primary_key_from_returning_getter
2316 compiled_params = self.compiled_parameters
2318 return [
2319 getter(row, param) for row, param in zip(rows, compiled_params)
2320 ]
2322 def lastrow_has_defaults(self) -> bool:
2323 return (self.isinsert or self.isupdate) and bool(
2324 cast(SQLCompiler, self.compiled).postfetch
2325 )
2327 def _prepare_set_input_sizes(
2328 self,
2329 ) -> Optional[List[Tuple[str, Any, TypeEngine[Any]]]]:
2330 """Given a cursor and ClauseParameters, prepare arguments
2331 in order to call the appropriate
2332 style of ``setinputsizes()`` on the cursor, using DB-API types
2333 from the bind parameter's ``TypeEngine`` objects.
2335 This method only called by those dialects which set the
2336 :attr:`.Dialect.bind_typing` attribute to
2337 :attr:`.BindTyping.SETINPUTSIZES`. Python-oracledb and cx_Oracle are
2338 the only DBAPIs that requires setinputsizes(); pyodbc offers it as an
2339 option.
2341 Prior to SQLAlchemy 2.0, the setinputsizes() approach was also used
2342 for pg8000 and asyncpg, which has been changed to inline rendering
2343 of casts.
2345 """
2346 if self.isddl or self.is_text:
2347 return None
2349 compiled = cast(SQLCompiler, self.compiled)
2351 inputsizes = compiled._get_set_input_sizes_lookup()
2353 if inputsizes is None:
2354 return None
2356 dialect = self.dialect
2358 # all of the rest of this... cython?
2360 if dialect._has_events:
2361 inputsizes = dict(inputsizes)
2362 dialect.dispatch.do_setinputsizes(
2363 inputsizes, self.cursor, self.statement, self.parameters, self
2364 )
2366 if compiled.escaped_bind_names:
2367 escaped_bind_names = compiled.escaped_bind_names
2368 else:
2369 escaped_bind_names = None
2371 if dialect.positional:
2372 items = [
2373 (key, compiled.binds[key])
2374 for key in compiled.positiontup or ()
2375 ]
2376 else:
2377 items = [
2378 (key, bindparam)
2379 for bindparam, key in compiled.bind_names.items()
2380 ]
2382 generic_inputsizes: List[Tuple[str, Any, TypeEngine[Any]]] = []
2383 for key, bindparam in items:
2384 if bindparam in compiled.literal_execute_params:
2385 continue
2387 if key in self._expanded_parameters:
2388 if is_tuple_type(bindparam.type):
2389 num = len(bindparam.type.types)
2390 dbtypes = inputsizes[bindparam]
2391 generic_inputsizes.extend(
2392 (
2393 (
2394 escaped_bind_names.get(paramname, paramname)
2395 if escaped_bind_names is not None
2396 else paramname
2397 ),
2398 dbtypes[idx % num],
2399 bindparam.type.types[idx % num],
2400 )
2401 for idx, paramname in enumerate(
2402 self._expanded_parameters[key]
2403 )
2404 )
2405 else:
2406 dbtype = inputsizes.get(bindparam, None)
2407 generic_inputsizes.extend(
2408 (
2409 (
2410 escaped_bind_names.get(paramname, paramname)
2411 if escaped_bind_names is not None
2412 else paramname
2413 ),
2414 dbtype,
2415 bindparam.type,
2416 )
2417 for paramname in self._expanded_parameters[key]
2418 )
2419 else:
2420 dbtype = inputsizes.get(bindparam, None)
2422 escaped_name = (
2423 escaped_bind_names.get(key, key)
2424 if escaped_bind_names is not None
2425 else key
2426 )
2428 generic_inputsizes.append(
2429 (escaped_name, dbtype, bindparam.type)
2430 )
2432 return generic_inputsizes
2434 def _exec_default(self, column, default, type_):
2435 if default.is_sequence:
2436 return self.fire_sequence(default, type_)
2437 elif default.is_callable:
2438 # this codepath is not normally used as it's inlined
2439 # into _process_execute_defaults
2440 self.current_column = column
2441 return default.arg(self)
2442 elif default.is_clause_element:
2443 return self._exec_default_clause_element(column, default, type_)
2444 else:
2445 # this codepath is not normally used as it's inlined
2446 # into _process_execute_defaults
2447 return default.arg
2449 def _exec_default_clause_element(self, column, default, type_):
2450 # execute a default that's a complete clause element. Here, we have
2451 # to re-implement a miniature version of the compile->parameters->
2452 # cursor.execute() sequence, since we don't want to modify the state
2453 # of the connection / result in progress or create new connection/
2454 # result objects etc.
2455 # .. versionchanged:: 1.4
2457 if not default._arg_is_typed:
2458 default_arg = expression.type_coerce(default.arg, type_)
2459 else:
2460 default_arg = default.arg
2461 compiled = expression.select(default_arg).compile(dialect=self.dialect)
2462 compiled_params = compiled.construct_params()
2463 processors = compiled._bind_processors
2464 if compiled.positional:
2465 parameters = self.dialect.execute_sequence_format(
2466 [
2467 (
2468 processors[key](compiled_params[key]) # type: ignore[operator] # noqa: E501
2469 if key in processors
2470 else compiled_params[key]
2471 )
2472 for key in compiled.positiontup or ()
2473 ]
2474 )
2475 else:
2476 parameters = {
2477 key: (
2478 processors[key](compiled_params[key]) # type: ignore[assignment, operator] # noqa: E501
2479 if key in processors
2480 else compiled_params[key]
2481 )
2482 for key in compiled_params
2483 }
2484 return self._execute_scalar(
2485 str(compiled), type_, parameters=parameters
2486 )
2488 current_parameters: Optional[_CoreSingleExecuteParams] = None
2489 """A dictionary of parameters applied to the current row.
2491 This attribute is only available in the context of a user-defined default
2492 generation function, e.g. as described at :ref:`context_default_functions`.
2493 It consists of a dictionary which includes entries for each column/value
2494 pair that is to be part of the INSERT or UPDATE statement. The keys of the
2495 dictionary will be the key value of each :class:`_schema.Column`,
2496 which is usually
2497 synonymous with the name.
2499 Note that the :attr:`.DefaultExecutionContext.current_parameters` attribute
2500 does not accommodate for the "multi-values" feature of the
2501 :meth:`_expression.Insert.values` method. The
2502 :meth:`.DefaultExecutionContext.get_current_parameters` method should be
2503 preferred.
2505 .. seealso::
2507 :meth:`.DefaultExecutionContext.get_current_parameters`
2509 :ref:`context_default_functions`
2511 """
2513 def get_current_parameters(self, isolate_multiinsert_groups=True):
2514 """Return a dictionary of parameters applied to the current row.
2516 This method can only be used in the context of a user-defined default
2517 generation function, e.g. as described at
2518 :ref:`context_default_functions`. When invoked, a dictionary is
2519 returned which includes entries for each column/value pair that is part
2520 of the INSERT or UPDATE statement. The keys of the dictionary will be
2521 the key value of each :class:`_schema.Column`,
2522 which is usually synonymous
2523 with the name.
2525 :param isolate_multiinsert_groups=True: indicates that multi-valued
2526 INSERT constructs created using :meth:`_expression.Insert.values`
2527 should be
2528 handled by returning only the subset of parameters that are local
2529 to the current column default invocation. When ``False``, the
2530 raw parameters of the statement are returned including the
2531 naming convention used in the case of multi-valued INSERT.
2533 .. seealso::
2535 :attr:`.DefaultExecutionContext.current_parameters`
2537 :ref:`context_default_functions`
2539 """
2540 try:
2541 parameters = self.current_parameters
2542 column = self.current_column
2543 except AttributeError:
2544 raise exc.InvalidRequestError(
2545 "get_current_parameters() can only be invoked in the "
2546 "context of a Python side column default function"
2547 )
2548 else:
2549 assert column is not None
2550 assert parameters is not None
2551 compile_state = cast(
2552 "DMLState", cast(SQLCompiler, self.compiled).compile_state
2553 )
2554 assert compile_state is not None
2555 if (
2556 isolate_multiinsert_groups
2557 and dml.isinsert(compile_state)
2558 and compile_state._has_multi_parameters
2559 ):
2560 if column._is_multiparam_column:
2561 index = column.index + 1
2562 d = {column.original.key: parameters[column.key]}
2563 else:
2564 d = {column.key: parameters[column.key]}
2565 index = 0
2566 assert compile_state._dict_parameters is not None
2567 keys = compile_state._dict_parameters.keys()
2568 d.update(
2569 (key, parameters["%s_m%d" % (key, index)]) for key in keys
2570 )
2571 return d
2572 else:
2573 return parameters
2575 def get_insert_default(self, column):
2576 if column.default is None:
2577 return None
2578 else:
2579 return self._exec_default(column, column.default, column.type)
2581 def get_update_default(self, column):
2582 if column.onupdate is None:
2583 return None
2584 else:
2585 return self._exec_default(column, column.onupdate, column.type)
2587 def _process_execute_defaults(self):
2588 compiled = cast(SQLCompiler, self.compiled)
2590 key_getter = compiled._within_exec_param_key_getter
2592 sentinel_counter = 0
2594 if compiled.insert_prefetch:
2595 prefetch_recs = [
2596 (
2597 c,
2598 key_getter(c),
2599 c._default_description_tuple,
2600 self.get_insert_default,
2601 )
2602 for c in compiled.insert_prefetch
2603 ]
2604 elif compiled.update_prefetch:
2605 prefetch_recs = [
2606 (
2607 c,
2608 key_getter(c),
2609 c._onupdate_description_tuple,
2610 self.get_update_default,
2611 )
2612 for c in compiled.update_prefetch
2613 ]
2614 else:
2615 prefetch_recs = []
2617 for param in self.compiled_parameters:
2618 self.current_parameters = param
2620 for (
2621 c,
2622 param_key,
2623 (arg, is_scalar, is_callable, is_sentinel),
2624 fallback,
2625 ) in prefetch_recs:
2626 if is_sentinel:
2627 param[param_key] = sentinel_counter
2628 sentinel_counter += 1
2629 elif is_scalar:
2630 param[param_key] = arg
2631 elif is_callable:
2632 self.current_column = c
2633 param[param_key] = arg(self)
2634 else:
2635 val = fallback(c)
2636 if val is not None:
2637 param[param_key] = val
2639 del self.current_parameters
2642DefaultDialect.execution_ctx_cls = DefaultExecutionContext