Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py: 48%

1# engine/base.py

3# <see AUTHORS file>

5# This module is part of SQLAlchemy and is released under

6# the MIT License: https://www.opensource.org/licenses/mit-license.php

7"""Defines :class:`_engine.Connection` and :class:`_engine.Engine`."""

9from __future__ import annotations

11import contextlib

12import sys

13import typing

14from typing import Any

15from typing import Callable

16from typing import cast

17from typing import Iterable

18from typing import Iterator

19from typing import List

20from typing import Mapping

21from typing import NoReturn

22from typing import Optional

23from typing import overload

24from typing import Tuple

25from typing import Type

26from typing import TypeVar

27from typing import Union

29from .interfaces import BindTyping

30from .interfaces import ConnectionEventsTarget

31from .interfaces import DBAPICursor

32from .interfaces import ExceptionContext

33from .interfaces import ExecuteStyle

34from .interfaces import ExecutionContext

35from .interfaces import IsolationLevel

36from .util import _distill_params_20

37from .util import _distill_raw_params

38from .util import TransactionalContext

39from .. import exc

40from .. import inspection

41from .. import log

42from .. import util

43from ..sql import compiler

44from ..sql import util as sql_util

45from ..util.typing import Never

46from ..util.typing import TupleAny

47from ..util.typing import TypeVarTuple

48from ..util.typing import Unpack

50if typing.TYPE_CHECKING:

51 from . import CursorResult

52 from . import ScalarResult

53 from .interfaces import _AnyExecuteParams

54 from .interfaces import _AnyMultiExecuteParams

55 from .interfaces import _CoreAnyExecuteParams

56 from .interfaces import _CoreMultiExecuteParams

57 from .interfaces import _CoreSingleExecuteParams

58 from .interfaces import _DBAPIAnyExecuteParams

59 from .interfaces import _DBAPISingleExecuteParams

60 from .interfaces import _ExecuteOptions

61 from .interfaces import CompiledCacheType

62 from .interfaces import CoreExecuteOptionsParameter

63 from .interfaces import Dialect

64 from .interfaces import SchemaTranslateMapType

65 from .reflection import Inspector # noqa

66 from .url import URL

67 from ..event import dispatcher

68 from ..log import _EchoFlagType

69 from ..pool import _ConnectionFairy

70 from ..pool import Pool

71 from ..pool import PoolProxiedConnection

72 from ..sql import Executable

73 from ..sql._typing import _InfoType

74 from ..sql.compiler import Compiled

75 from ..sql.ddl import ExecutableDDLElement

76 from ..sql.ddl import InvokeDDLBase

77 from ..sql.functions import FunctionElement

78 from ..sql.schema import DefaultGenerator

79 from ..sql.schema import HasSchemaAttr

80 from ..sql.schema import SchemaVisitable

81 from ..sql.selectable import TypedReturnsRows

84_T = TypeVar("_T", bound=Any)

85_Ts = TypeVarTuple("_Ts")

86_EMPTY_EXECUTION_OPTS: _ExecuteOptions = util.EMPTY_DICT

87NO_OPTIONS: Mapping[str, Any] = util.EMPTY_DICT

90class Connection(ConnectionEventsTarget, inspection.Inspectable["Inspector"]):

91 """Provides high-level functionality for a wrapped DB-API connection.

93 The :class:`_engine.Connection` object is procured by calling the

94 :meth:`_engine.Engine.connect` method of the :class:`_engine.Engine`

95 object, and provides services for execution of SQL statements as well

96 as transaction control.

98 The Connection object is **not** thread-safe. While a Connection can be

99 shared among threads using properly synchronized access, it is still

100 possible that the underlying DBAPI connection may not support shared

101 access between threads. Check the DBAPI documentation for details.

102

103 The Connection object represents a single DBAPI connection checked out

104 from the connection pool. In this state, the connection pool has no

105 affect upon the connection, including its expiration or timeout state.

106 For the connection pool to properly manage connections, connections

107 should be returned to the connection pool (i.e. ``connection.close()``)

108 whenever the connection is not in use.

109

110 .. index::

111 single: thread safety; Connection

112

113 """

114

115 dialect: Dialect

116 dispatch: dispatcher[ConnectionEventsTarget]

117

118 _sqla_logger_namespace = "sqlalchemy.engine.Connection"

119

120 # used by sqlalchemy.engine.util.TransactionalContext

121 _trans_context_manager: Optional[TransactionalContext] = None

122

123 # legacy as of 2.0, should be eventually deprecated and

124 # removed. was used in the "pre_ping" recipe that's been in the docs

125 # a long time

126 should_close_with_result = False

127

128 _dbapi_connection: Optional[PoolProxiedConnection]

129

130 _execution_options: _ExecuteOptions

131

132 _transaction: Optional[RootTransaction]

133 _nested_transaction: Optional[NestedTransaction]

134

135 def __init__(

136 self,

137 engine: Engine,

138 connection: Optional[PoolProxiedConnection] = None,

139 _has_events: Optional[bool] = None,

140 _allow_revalidate: bool = True,

141 _allow_autobegin: bool = True,

142 ):

143 """Construct a new Connection."""

144 self.engine = engine

145 self.dialect = dialect = engine.dialect

146

147 if connection is None:

148 try:

149 self._dbapi_connection = engine.raw_connection()

150 except dialect.loaded_dbapi.Error as err:

151 Connection._handle_dbapi_exception_noconnection(

152 err, dialect, engine

153 )

154 raise

155 else:

156 self._dbapi_connection = connection

157

158 self._transaction = self._nested_transaction = None

159 self.__savepoint_seq = 0

160 self.__in_begin = False

161

162 self.__can_reconnect = _allow_revalidate

163 self._allow_autobegin = _allow_autobegin

164 self._echo = self.engine._should_log_info()

165

166 if _has_events is None:

167 # if _has_events is sent explicitly as False,

168 # then don't join the dispatch of the engine; we don't

169 # want to handle any of the engine's events in that case.

170 self.dispatch = self.dispatch._join(engine.dispatch)

171 self._has_events = _has_events or (

172 _has_events is None and engine._has_events

173 )

174

175 self._execution_options = engine._execution_options

176

177 if self._has_events or self.engine._has_events:

178 self.dispatch.engine_connect(self)

179

180 # this can be assigned differently via

181 # characteristics.LoggingTokenCharacteristic

182 _message_formatter: Any = None

183

184 def _log_parameters(

185 self,

186 parameters: _AnyExecuteParams | None,

187 stats: str,

188 *,

189 ismulti: bool,

190 ) -> None:

191 if not self.engine.hide_parameters:

192 self._log_info(

193 "[%s] %r",

194 stats,

195 sql_util._repr_params(

196 parameters,

197 batches=10,

198 ismulti=ismulti,

199 ),

200 )

201 else:

202 self._log_info(

203 "[%s] [SQL parameters hidden due to hide_parameters=True]",

204 stats,

205 )

206

207 def _log_info(self, message: str, *arg: Any, **kw: Any) -> None:

208 fmt = self._message_formatter

209

210 if fmt:

211 message = fmt(message)

212

213 if log.STACKLEVEL:

214 kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET

215

216 self.engine.logger.info(message, *arg, **kw)

217

218 def _log_debug(self, message: str, *arg: Any, **kw: Any) -> None:

219 fmt = self._message_formatter

220

221 if fmt:

222 message = fmt(message)

223

224 if log.STACKLEVEL:

225 kw["stacklevel"] = 1 + log.STACKLEVEL_OFFSET

226

227 self.engine.logger.debug(message, *arg, **kw)

228

229 @property

230 def _schema_translate_map(self) -> Optional[SchemaTranslateMapType]:

231 schema_translate_map: Optional[SchemaTranslateMapType] = (

232 self._execution_options.get("schema_translate_map", None)

233 )

234

235 return schema_translate_map

236

237 def schema_for_object(self, obj: HasSchemaAttr) -> Optional[str]:

238 """Return the schema name for the given schema item taking into

239 account current schema translate map.

240

241 """

242

243 name = obj.schema

244 schema_translate_map: Optional[SchemaTranslateMapType] = (

245 self._execution_options.get("schema_translate_map", None)

246 )

247

248 if (

249 schema_translate_map

250 and name in schema_translate_map

251 and obj._use_schema_map

252 ):

253 return schema_translate_map[name]

254 else:

255 return name

256

257 def __enter__(self) -> Connection:

258 return self

259

260 def __exit__(self, type_: Any, value: Any, traceback: Any) -> None:

261 self.close()

262

263 @overload

264 def execution_options(

265 self,

266 *,

267 compiled_cache: Optional[CompiledCacheType] = ...,

268 logging_token: str = ...,

269 isolation_level: IsolationLevel = ...,

270 no_parameters: bool = False,

271 stream_results: bool = False,

272 max_row_buffer: int = ...,

273 yield_per: int = ...,

274 insertmanyvalues_page_size: int = ...,

275 schema_translate_map: Optional[SchemaTranslateMapType] = ...,

276 preserve_rowcount: bool = False,

277 driver_column_names: bool = False,

278 **opt: Any,

279 ) -> Connection: ...

280

281 @overload

282 def execution_options(self, **opt: Any) -> Connection: ...

283

284 def execution_options(self, **opt: Any) -> Connection:

285 r"""Set non-SQL options for the connection which take effect

286 during execution.

287

288 This method modifies this :class:`_engine.Connection` **in-place**;

289 the return value is the same :class:`_engine.Connection` object

290 upon which the method is called. Note that this is in contrast

291 to the behavior of the ``execution_options`` methods on other

292 objects such as :meth:`_engine.Engine.execution_options` and

293 :meth:`_sql.Executable.execution_options`. The rationale is that many

294 such execution options necessarily modify the state of the base

295 DBAPI connection in any case so there is no feasible means of

296 keeping the effect of such an option localized to a "sub" connection.

297

298 .. versionchanged:: 2.0 The :meth:`_engine.Connection.execution_options`

299 method, in contrast to other objects with this method, modifies

300 the connection in-place without creating copy of it.

301

302 As discussed elsewhere, the :meth:`_engine.Connection.execution_options`

303 method accepts any arbitrary parameters including user defined names.

304 All parameters given are consumable in a number of ways including

305 by using the :meth:`_engine.Connection.get_execution_options` method.

306 See the examples at :meth:`_sql.Executable.execution_options`

307 and :meth:`_engine.Engine.execution_options`.

308

309 The keywords that are currently recognized by SQLAlchemy itself

310 include all those listed under :meth:`.Executable.execution_options`,

311 as well as others that are specific to :class:`_engine.Connection`.

312

313 :param compiled_cache: Available on: :class:`_engine.Connection`,

314 :class:`_engine.Engine`.

315

316 A dictionary where :class:`.Compiled` objects

317 will be cached when the :class:`_engine.Connection`

318 compiles a clause

319 expression into a :class:`.Compiled` object. This dictionary will

320 supersede the statement cache that may be configured on the

321 :class:`_engine.Engine` itself. If set to None, caching

322 is disabled, even if the engine has a configured cache size.

323

324 Note that the ORM makes use of its own "compiled" caches for

325 some operations, including flush operations. The caching

326 used by the ORM internally supersedes a cache dictionary

327 specified here.

328

329 :param logging_token: Available on: :class:`_engine.Connection`,

330 :class:`_engine.Engine`, :class:`_sql.Executable`.

331

332 Adds the specified string token surrounded by brackets in log

333 messages logged by the connection, i.e. the logging that's enabled

334 either via the :paramref:`_sa.create_engine.echo` flag or via the

335 ``logging.getLogger("sqlalchemy.engine")`` logger. This allows a

336 per-connection or per-sub-engine token to be available which is

337 useful for debugging concurrent connection scenarios.

338

339 .. versionadded:: 1.4.0b2

340

341 .. seealso::

342

343 :ref:`dbengine_logging_tokens` - usage example

344

345 :paramref:`_sa.create_engine.logging_name` - adds a name to the

346 name used by the Python logger object itself.

347

348 :param isolation_level: Available on: :class:`_engine.Connection`,

349 :class:`_engine.Engine`.

350

351 Set the transaction isolation level for the lifespan of this

352 :class:`_engine.Connection` object.

353 Valid values include those string

354 values accepted by the :paramref:`_sa.create_engine.isolation_level`

355 parameter passed to :func:`_sa.create_engine`. These levels are

356 semi-database specific; see individual dialect documentation for

357 valid levels.

358

359 The isolation level option applies the isolation level by emitting

360 statements on the DBAPI connection, and **necessarily affects the

361 original Connection object overall**. The isolation level will remain

362 at the given setting until explicitly changed, or when the DBAPI

363 connection itself is :term:`released` to the connection pool, i.e. the

364 :meth:`_engine.Connection.close` method is called, at which time an

365 event handler will emit additional statements on the DBAPI connection

366 in order to revert the isolation level change.

367

368 .. note:: The ``isolation_level`` execution option may only be

369 established before the :meth:`_engine.Connection.begin` method is

370 called, as well as before any SQL statements are emitted which

371 would otherwise trigger "autobegin", or directly after a call to

372 :meth:`_engine.Connection.commit` or

373 :meth:`_engine.Connection.rollback`. A database cannot change the

374 isolation level on a transaction in progress.

375

376 .. note:: The ``isolation_level`` execution option is implicitly

377 reset if the :class:`_engine.Connection` is invalidated, e.g. via

378 the :meth:`_engine.Connection.invalidate` method, or if a

379 disconnection error occurs. The new connection produced after the

380 invalidation will **not** have the selected isolation level

381 re-applied to it automatically.

382

383 .. seealso::

384

385 :ref:`dbapi_autocommit`

386

387 :meth:`_engine.Connection.get_isolation_level`

388 - view current actual level

389

390 :param no_parameters: Available on: :class:`_engine.Connection`,

391 :class:`_sql.Executable`.

392

393 When ``True``, if the final parameter

394 list or dictionary is totally empty, will invoke the

395 statement on the cursor as ``cursor.execute(statement)``,

396 not passing the parameter collection at all.

397 Some DBAPIs such as psycopg2 and mysql-python consider

398 percent signs as significant only when parameters are

399 present; this option allows code to generate SQL

400 containing percent signs (and possibly other characters)

401 that is neutral regarding whether it's executed by the DBAPI

402 or piped into a script that's later invoked by

403 command line tools.

404

405 :param stream_results: Available on: :class:`_engine.Connection`,

406 :class:`_sql.Executable`.

407

408 Indicate to the dialect that results should be "streamed" and not

409 pre-buffered, if possible. For backends such as PostgreSQL, MySQL

410 and MariaDB, this indicates the use of a "server side cursor" as

411 opposed to a client side cursor. Other backends such as that of

412 Oracle Database may already use server side cursors by default.

413

414 The usage of

415 :paramref:`_engine.Connection.execution_options.stream_results` is

416 usually combined with setting a fixed number of rows to to be fetched

417 in batches, to allow for efficient iteration of database rows while

418 at the same time not loading all result rows into memory at once;

419 this can be configured on a :class:`_engine.Result` object using the

420 :meth:`_engine.Result.yield_per` method, after execution has

421 returned a new :class:`_engine.Result`. If

422 :meth:`_engine.Result.yield_per` is not used,

423 the :paramref:`_engine.Connection.execution_options.stream_results`

424 mode of operation will instead use a dynamically sized buffer

425 which buffers sets of rows at a time, growing on each batch

426 based on a fixed growth size up until a limit which may

427 be configured using the

428 :paramref:`_engine.Connection.execution_options.max_row_buffer`

429 parameter.

430

431 When using the ORM to fetch ORM mapped objects from a result,

432 :meth:`_engine.Result.yield_per` should always be used with

433 :paramref:`_engine.Connection.execution_options.stream_results`,

434 so that the ORM does not fetch all rows into new ORM objects at once.

435

436 For typical use, the

437 :paramref:`_engine.Connection.execution_options.yield_per` execution

438 option should be preferred, which sets up both

439 :paramref:`_engine.Connection.execution_options.stream_results` and

440 :meth:`_engine.Result.yield_per` at once. This option is supported

441 both at a core level by :class:`_engine.Connection` as well as by the

442 ORM :class:`_engine.Session`; the latter is described at

443 :ref:`orm_queryguide_yield_per`.

444

445 .. seealso::

446

447 :ref:`engine_stream_results` - background on

448 :paramref:`_engine.Connection.execution_options.stream_results`

449

450 :paramref:`_engine.Connection.execution_options.max_row_buffer`

451

452 :paramref:`_engine.Connection.execution_options.yield_per`

453

454 :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`

455 describing the ORM version of ``yield_per``

456

457 :param max_row_buffer: Available on: :class:`_engine.Connection`,

458 :class:`_sql.Executable`. Sets a maximum

459 buffer size to use when the

460 :paramref:`_engine.Connection.execution_options.stream_results`

461 execution option is used on a backend that supports server side

462 cursors. The default value if not specified is 1000.

463

464 .. seealso::

465

466 :paramref:`_engine.Connection.execution_options.stream_results`

467

468 :ref:`engine_stream_results`

469

470

471 :param yield_per: Available on: :class:`_engine.Connection`,

472 :class:`_sql.Executable`. Integer value applied which will

473 set the :paramref:`_engine.Connection.execution_options.stream_results`

474 execution option and invoke :meth:`_engine.Result.yield_per`

475 automatically at once. Allows equivalent functionality as

476 is present when using this parameter with the ORM.

477

478 .. versionadded:: 1.4.40

479

480 .. seealso::

481

482 :ref:`engine_stream_results` - background and examples

483 on using server side cursors with Core.

484

485 :ref:`orm_queryguide_yield_per` - in the :ref:`queryguide_toplevel`

486 describing the ORM version of ``yield_per``

487

488 :param insertmanyvalues_page_size: Available on: :class:`_engine.Connection`,

489 :class:`_engine.Engine`. Number of rows to format into an

490 INSERT statement when the statement uses "insertmanyvalues" mode,

491 which is a paged form of bulk insert that is used for many backends

492 when using :term:`executemany` execution typically in conjunction

493 with RETURNING. Defaults to 1000. May also be modified on a

494 per-engine basis using the

495 :paramref:`_sa.create_engine.insertmanyvalues_page_size` parameter.

496

497 .. versionadded:: 2.0

498

499 .. seealso::

500

501 :ref:`engine_insertmanyvalues`

502

503 :param schema_translate_map: Available on: :class:`_engine.Connection`,

504 :class:`_engine.Engine`, :class:`_sql.Executable`.

505

506 A dictionary mapping schema names to schema names, that will be

507 applied to the :paramref:`_schema.Table.schema` element of each

508 :class:`_schema.Table`

509 encountered when SQL or DDL expression elements

510 are compiled into strings; the resulting schema name will be

511 converted based on presence in the map of the original name.

512

513 .. seealso::

514

515 :ref:`schema_translating`

516

517 :param preserve_rowcount: Boolean; when True, the ``cursor.rowcount``

518 attribute will be unconditionally memoized within the result and

519 made available via the :attr:`.CursorResult.rowcount` attribute.

520 Normally, this attribute is only preserved for UPDATE and DELETE

521 statements. Using this option, the DBAPIs rowcount value can

522 be accessed for other kinds of statements such as INSERT and SELECT,

523 to the degree that the DBAPI supports these statements. See

524 :attr:`.CursorResult.rowcount` for notes regarding the behavior

525 of this attribute.

526

527 .. versionadded:: 2.0.28

528

529 .. seealso::

530

531 :meth:`_engine.Engine.execution_options`

532

533 :meth:`.Executable.execution_options`

534

535 :meth:`_engine.Connection.get_execution_options`

536

537 :ref:`orm_queryguide_execution_options` - documentation on all

538 ORM-specific execution options

539

540 :param driver_column_names: When True, the returned

541 :class:`_engine.CursorResult` will use the column names as written in

542 ``cursor.description`` to set up the keys for the result set,

543 including the names of columns for the :class:`_engine.Row` object as

544 well as the dictionary keys when using :attr:`_engine.Row._mapping`.

545 On backends that use "name normalization" such as Oracle Database to

546 correct for lower case names being converted to all uppercase, this

547 behavior is turned off and the raw UPPERCASE names in

548 cursor.description will be present.

549

550 .. versionadded:: 2.1

551

552 """ # noqa

553 if self._has_events or self.engine._has_events:

554 self.dispatch.set_connection_execution_options(self, opt)

555 self._execution_options = self._execution_options.union(opt)

556 self.dialect.set_connection_execution_options(self, opt)

557 return self

558

559 def get_execution_options(self) -> _ExecuteOptions:

560 """Get the non-SQL options which will take effect during execution.

561

562 .. seealso::

563

564 :meth:`_engine.Connection.execution_options`

565 """

566 return self._execution_options

567

568 @property

569 def _still_open_and_dbapi_connection_is_valid(self) -> bool:

570 pool_proxied_connection = self._dbapi_connection

571 return (

572 pool_proxied_connection is not None

573 and pool_proxied_connection.is_valid

574 )

575

576 @property

577 def closed(self) -> bool:

578 """Return True if this connection is closed."""

579

580 return self._dbapi_connection is None and not self.__can_reconnect

581

582 @property

583 def invalidated(self) -> bool:

584 """Return True if this connection was invalidated.

585

586 This does not indicate whether or not the connection was

587 invalidated at the pool level, however

588

589 """

590

591 # prior to 1.4, "invalid" was stored as a state independent of

592 # "closed", meaning an invalidated connection could be "closed",

593 # the _dbapi_connection would be None and closed=True, yet the

594 # "invalid" flag would stay True. This meant that there were

595 # three separate states (open/valid, closed/valid, closed/invalid)

596 # when there is really no reason for that; a connection that's

597 # "closed" does not need to be "invalid". So the state is now

598 # represented by the two facts alone.

599

600 pool_proxied_connection = self._dbapi_connection

601 return pool_proxied_connection is None and self.__can_reconnect

602

603 @property

604 def connection(self) -> PoolProxiedConnection:

605 """The underlying DB-API connection managed by this Connection.

606

607 This is a SQLAlchemy connection-pool proxied connection

608 which then has the attribute

609 :attr:`_pool._ConnectionFairy.dbapi_connection` that refers to the

610 actual driver connection.

611

612 .. seealso::

613

614

615 :ref:`dbapi_connections`

616

617 """

618

619 if self._dbapi_connection is None:

620 try:

621 return self._revalidate_connection()

622 except (exc.PendingRollbackError, exc.ResourceClosedError):

623 raise

624 except BaseException as e:

625 self._handle_dbapi_exception(e, None, None, None, None)

626 else:

627 return self._dbapi_connection

628

629 def get_isolation_level(self) -> IsolationLevel:

630 """Return the current **actual** isolation level that's present on

631 the database within the scope of this connection.

632

633 This attribute will perform a live SQL operation against the database

634 in order to procure the current isolation level, so the value returned

635 is the actual level on the underlying DBAPI connection regardless of

636 how this state was set. This will be one of the four actual isolation

637 modes ``READ UNCOMMITTED``, ``READ COMMITTED``, ``REPEATABLE READ``,

638 ``SERIALIZABLE``. It will **not** include the ``AUTOCOMMIT`` isolation

639 level setting. Third party dialects may also feature additional

640 isolation level settings.

641

642 .. note:: This method **will not report** on the ``AUTOCOMMIT``

643 isolation level, which is a separate :term:`dbapi` setting that's

644 independent of **actual** isolation level. When ``AUTOCOMMIT`` is

645 in use, the database connection still has a "traditional" isolation

646 mode in effect, that is typically one of the four values

647 ``READ UNCOMMITTED``, ``READ COMMITTED``, ``REPEATABLE READ``,

648 ``SERIALIZABLE``.

649

650 Compare to the :attr:`_engine.Connection.default_isolation_level`

651 accessor which returns the isolation level that is present on the

652 database at initial connection time.

653

654 .. seealso::

655

656 :attr:`_engine.Connection.default_isolation_level`

657 - view default level

658

659 :paramref:`_sa.create_engine.isolation_level`

660 - set per :class:`_engine.Engine` isolation level

661

662 :paramref:`.Connection.execution_options.isolation_level`

663 - set per :class:`_engine.Connection` isolation level

664

665 """

666 dbapi_connection = self.connection.dbapi_connection

667 assert dbapi_connection is not None

668 try:

669 return self.dialect.get_isolation_level(dbapi_connection)

670 except BaseException as e:

671 self._handle_dbapi_exception(e, None, None, None, None)

672

673 @property

674 def default_isolation_level(self) -> Optional[IsolationLevel]:

675 """The initial-connection time isolation level associated with the

676 :class:`_engine.Dialect` in use.

677

678 This value is independent of the

679 :paramref:`.Connection.execution_options.isolation_level` and

680 :paramref:`.Engine.execution_options.isolation_level` execution

681 options, and is determined by the :class:`_engine.Dialect` when the

682 first connection is created, by performing a SQL query against the

683 database for the current isolation level before any additional commands

684 have been emitted.

685

686 Calling this accessor does not invoke any new SQL queries.

687

688 .. seealso::

689

690 :meth:`_engine.Connection.get_isolation_level`

691 - view current actual isolation level

692

693 :paramref:`_sa.create_engine.isolation_level`

694 - set per :class:`_engine.Engine` isolation level

695

696 :paramref:`.Connection.execution_options.isolation_level`

697 - set per :class:`_engine.Connection` isolation level

698

699 """

700 return self.dialect.default_isolation_level

701

702 def _invalid_transaction(self) -> NoReturn:

703 raise exc.PendingRollbackError(

704 "Can't reconnect until invalid %stransaction is rolled "

705 "back. Please rollback() fully before proceeding"

706 % ("savepoint " if self._nested_transaction is not None else ""),

707 code="8s2b",

708 )

709

710 def _revalidate_connection(self) -> PoolProxiedConnection:

711 if self.__can_reconnect and self.invalidated:

712 if self._transaction is not None:

713 self._invalid_transaction()

714 self._dbapi_connection = self.engine.raw_connection()

715 return self._dbapi_connection

716 raise exc.ResourceClosedError("This Connection is closed")

717

718 @property

719 def info(self) -> _InfoType:

720 """Info dictionary associated with the underlying DBAPI connection

721 referred to by this :class:`_engine.Connection`, allowing user-defined

722 data to be associated with the connection.

723

724 The data here will follow along with the DBAPI connection including

725 after it is returned to the connection pool and used again

726 in subsequent instances of :class:`_engine.Connection`.

727

728 """

729

730 return self.connection.info

731

732 def invalidate(self, exception: Optional[BaseException] = None) -> None:

733 """Invalidate the underlying DBAPI connection associated with

734 this :class:`_engine.Connection`.

735

736 An attempt will be made to close the underlying DBAPI connection

737 immediately; however if this operation fails, the error is logged

738 but not raised. The connection is then discarded whether or not

739 close() succeeded.

740

741 Upon the next use (where "use" typically means using the

742 :meth:`_engine.Connection.execute` method or similar),

743 this :class:`_engine.Connection` will attempt to

744 procure a new DBAPI connection using the services of the

745 :class:`_pool.Pool` as a source of connectivity (e.g.

746 a "reconnection").

747

748 If a transaction was in progress (e.g. the

749 :meth:`_engine.Connection.begin` method has been called) when

750 :meth:`_engine.Connection.invalidate` method is called, at the DBAPI

751 level all state associated with this transaction is lost, as

752 the DBAPI connection is closed. The :class:`_engine.Connection`

753 will not allow a reconnection to proceed until the

754 :class:`.Transaction` object is ended, by calling the

755 :meth:`.Transaction.rollback` method; until that point, any attempt at

756 continuing to use the :class:`_engine.Connection` will raise an

757 :class:`~sqlalchemy.exc.InvalidRequestError`.

758 This is to prevent applications from accidentally

759 continuing an ongoing transactional operations despite the

760 fact that the transaction has been lost due to an

761 invalidation.

762

763 The :meth:`_engine.Connection.invalidate` method,

764 just like auto-invalidation,

765 will at the connection pool level invoke the

766 :meth:`_events.PoolEvents.invalidate` event.

767

768 :param exception: an optional ``Exception`` instance that's the

769 reason for the invalidation. is passed along to event handlers

770 and logging functions.

771

772 .. seealso::

773

774 :ref:`pool_connection_invalidation`

775

776 """

777

778 if self.invalidated:

779 return

780

781 if self.closed:

782 raise exc.ResourceClosedError("This Connection is closed")

783

784 try:

785 if self._still_open_and_dbapi_connection_is_valid:

786 pool_proxied_connection = self._dbapi_connection

787 assert pool_proxied_connection is not None

788 pool_proxied_connection.invalidate(exception)

789 finally:

790 # this is always a hard invalidation, which closes the DBAPI

791 # connection; for an asyncio driver that close is an awaited

792 # terminate() which a cancellation can interrupt. this

793 # Connection is invalidated regardless, so that nothing

794 # further, such as a rollback, is emitted on the DBAPI

795 # connection

796 self._dbapi_connection = None

797

798 def detach(self) -> None:

799 """Detach the underlying DB-API connection from its connection pool.

800

801 E.g.::

802

803 with engine.connect() as conn:

804 conn.detach()

805 conn.execute(text("SET search_path TO schema1, schema2"))

806

807 # work with connection

808

809 # connection is fully closed (since we used "with:", can

810 # also call .close())

811

812 This :class:`_engine.Connection` instance will remain usable.

813 When closed

814 (or exited from a context manager context as above),

815 the DB-API connection will be literally closed and not

816 returned to its originating pool.

817

818 This method can be used to insulate the rest of an application

819 from a modified state on a connection (such as a transaction

820 isolation level or similar).

821

822 """

823

824 if self.closed:

825 raise exc.ResourceClosedError("This Connection is closed")

826

827 pool_proxied_connection = self._dbapi_connection

828 if pool_proxied_connection is None:

829 raise exc.InvalidRequestError(

830 "Can't detach an invalidated Connection"

831 )

832 pool_proxied_connection.detach()

833

834 def _autobegin(self) -> None:

835 if self._allow_autobegin and not self.__in_begin:

836 self.begin()

837

838 def begin(self) -> RootTransaction:

839 """Begin a transaction prior to autobegin occurring.

840

841 E.g.::

842

843 with engine.connect() as conn:

844 with conn.begin() as trans:

845 conn.execute(table.insert(), {"username": "sandy"})

846

847 The returned object is an instance of :class:`_engine.RootTransaction`.

848 This object represents the "scope" of the transaction,

849 which completes when either the :meth:`_engine.Transaction.rollback`

850 or :meth:`_engine.Transaction.commit` method is called; the object

851 also works as a context manager as illustrated above.

852

853 The :meth:`_engine.Connection.begin` method begins a

854 transaction that normally will be begun in any case when the connection

855 is first used to execute a statement. The reason this method might be

856 used would be to invoke the :meth:`_events.ConnectionEvents.begin`

857 event at a specific time, or to organize code within the scope of a

858 connection checkout in terms of context managed blocks, such as::

859

860 with engine.connect() as conn:

861 with conn.begin():

862 conn.execute(...)

863 conn.execute(...)

864

865 with conn.begin():

866 conn.execute(...)

867 conn.execute(...)

868

869 The above code is not fundamentally any different in its behavior than

870 the following code which does not use

871 :meth:`_engine.Connection.begin`; the below style is known

872 as "commit as you go" style::

873

874 with engine.connect() as conn:

875 conn.execute(...)

876 conn.execute(...)

877 conn.commit()

878

879 conn.execute(...)

880 conn.execute(...)

881 conn.commit()

882

883 From a database point of view, the :meth:`_engine.Connection.begin`

884 method does not emit any SQL or change the state of the underlying

885 DBAPI connection in any way; the Python DBAPI does not have any

886 concept of explicit transaction begin.

887

888 .. seealso::

889

890 :ref:`tutorial_working_with_transactions` - in the

891 :ref:`unified_tutorial`

892

893 :meth:`_engine.Connection.begin_nested` - use a SAVEPOINT

894

895 :meth:`_engine.Connection.begin_twophase` -

896 use a two phase /XID transaction

897

898 :meth:`_engine.Engine.begin` - context manager available from

899 :class:`_engine.Engine`

900

901 """

902 if self._transaction is None:

903 self._transaction = RootTransaction(self)

904 return self._transaction

905 else:

906 raise exc.InvalidRequestError(

907 "This connection has already initialized a SQLAlchemy "

908 "Transaction() object via begin() or autobegin; can't "

909 "call begin() here unless rollback() or commit() "

910 "is called first."

911 )

912

913 def begin_nested(self) -> NestedTransaction:

914 """Begin a nested transaction (i.e. SAVEPOINT) and return a transaction

915 handle that controls the scope of the SAVEPOINT.

916

917 E.g.::

918

919 with engine.begin() as connection:

920 with connection.begin_nested():

921 connection.execute(table.insert(), {"username": "sandy"})

922

923 The returned object is an instance of

924 :class:`_engine.NestedTransaction`, which includes transactional

925 methods :meth:`_engine.NestedTransaction.commit` and

926 :meth:`_engine.NestedTransaction.rollback`; for a nested transaction,

927 these methods correspond to the operations "RELEASE SAVEPOINT <name>"

928 and "ROLLBACK TO SAVEPOINT <name>". The name of the savepoint is local

929 to the :class:`_engine.NestedTransaction` object and is generated

930 automatically. Like any other :class:`_engine.Transaction`, the

931 :class:`_engine.NestedTransaction` may be used as a context manager as

932 illustrated above which will "release" or "rollback" corresponding to

933 if the operation within the block were successful or raised an

934 exception.

935

936 Nested transactions require SAVEPOINT support in the underlying

937 database, else the behavior is undefined. SAVEPOINT is commonly used to

938 run operations within a transaction that may fail, while continuing the

939 outer transaction. E.g.::

940

941 from sqlalchemy import exc

942

943 with engine.begin() as connection:

944 trans = connection.begin_nested()

945 try:

946 connection.execute(table.insert(), {"username": "sandy"})

947 trans.commit()

948 except exc.IntegrityError: # catch for duplicate username

949 trans.rollback() # rollback to savepoint

950

951 # outer transaction continues

952 connection.execute(...)

953

954 If :meth:`_engine.Connection.begin_nested` is called without first

955 calling :meth:`_engine.Connection.begin` or

956 :meth:`_engine.Engine.begin`, the :class:`_engine.Connection` object

957 will "autobegin" the outer transaction first. This outer transaction

958 may be committed using "commit-as-you-go" style, e.g.::

959

960 with engine.connect() as connection: # begin() wasn't called

961

962 with connection.begin_nested(): # will auto-"begin()" first

963 connection.execute(...)

964 # savepoint is released

965

966 connection.execute(...)

967

968 # explicitly commit outer transaction

969 connection.commit()

970

971 # can continue working with connection here

972

973 .. versionchanged:: 2.0

974

975 :meth:`_engine.Connection.begin_nested` will now participate

976 in the connection "autobegin" behavior that is new as of

977 2.0 / "future" style connections in 1.4.

978

979 .. seealso::

980

981 :meth:`_engine.Connection.begin`

982

983 :ref:`session_begin_nested` - ORM support for SAVEPOINT

984

985 """

986 if self._transaction is None:

987 self._autobegin()

988

989 return NestedTransaction(self)

990

991 def begin_twophase(self, xid: Optional[Any] = None) -> TwoPhaseTransaction:

992 """Begin a two-phase or XA transaction and return a transaction

993 handle.

994

995 The returned object is an instance of :class:`.TwoPhaseTransaction`,

996 which in addition to the methods provided by

997 :class:`.Transaction`, also provides a

998 :meth:`~.TwoPhaseTransaction.prepare` method.

999

1000 :param xid: the two phase transaction id. If not supplied, a

1001 random id will be generated. The accepted type and value depends on

1002 the driver in use.

1003

1004 .. seealso::

1005

1006 :meth:`_engine.Connection.begin`

1007

1008 :meth:`_engine.Connection.begin_twophase`

1009

1010 """

1011

1012 if self._transaction is not None:

1013 raise exc.InvalidRequestError(

1014 "Cannot start a two phase transaction when a transaction "

1015 "is already in progress."

1016 )

1017 if xid is None:

1018 xid = self.engine.dialect.create_xid()

1019 return TwoPhaseTransaction(self, xid)

1020

1021 def commit(self) -> None:

1022 """Commit the transaction that is currently in progress.

1023

1024 This method commits the current transaction if one has been started.

1025 If no transaction was started, the method has no effect, assuming

1026 the connection is in a non-invalidated state.

1027

1028 A transaction is begun on a :class:`_engine.Connection` automatically

1029 whenever a statement is first executed, or when the

1030 :meth:`_engine.Connection.begin` method is called.

1031

1032 .. note:: The :meth:`_engine.Connection.commit` method only acts upon

1033 the primary database transaction that is linked to the

1034 :class:`_engine.Connection` object. It does not operate upon a

1035 SAVEPOINT that would have been invoked from the

1036 :meth:`_engine.Connection.begin_nested` method; for control of a

1037 SAVEPOINT, call :meth:`_engine.NestedTransaction.commit` on the

1038 :class:`_engine.NestedTransaction` that is returned by the

1039 :meth:`_engine.Connection.begin_nested` method itself.

1040

1041

1042 """

1043 if self._transaction:

1044 self._transaction.commit()

1045

1046 def rollback(self) -> None:

1047 """Roll back the transaction that is currently in progress.

1048

1049 This method rolls back the current transaction if one has been started.

1050 If no transaction was started, the method has no effect. If a

1051 transaction was started and the connection is in an invalidated state,

1052 the transaction is cleared using this method.

1053

1054 A transaction is begun on a :class:`_engine.Connection` automatically

1055 whenever a statement is first executed, or when the

1056 :meth:`_engine.Connection.begin` method is called.

1057

1058 .. note:: The :meth:`_engine.Connection.rollback` method only acts

1059 upon the primary database transaction that is linked to the

1060 :class:`_engine.Connection` object. It does not operate upon a

1061 SAVEPOINT that would have been invoked from the

1062 :meth:`_engine.Connection.begin_nested` method; for control of a

1063 SAVEPOINT, call :meth:`_engine.NestedTransaction.rollback` on the

1064 :class:`_engine.NestedTransaction` that is returned by the

1065 :meth:`_engine.Connection.begin_nested` method itself.

1066

1067

1068 """

1069 if self._transaction:

1070 self._transaction.rollback()

1071

1072 def recover_twophase(self) -> List[Any]:

1073 return self.engine.dialect.do_recover_twophase(self)

1074

1075 def rollback_prepared(self, xid: Any, recover: bool = False) -> None:

1076 self.engine.dialect.do_rollback_twophase(self, xid, recover=recover)

1077

1078 def commit_prepared(self, xid: Any, recover: bool = False) -> None:

1079 self.engine.dialect.do_commit_twophase(self, xid, recover=recover)

1080

1081 def in_transaction(self) -> bool:

1082 """Return True if a transaction is in progress."""

1083 return self._transaction is not None and self._transaction.is_active

1084

1085 def in_nested_transaction(self) -> bool:

1086 """Return True if a transaction is in progress."""

1087 return (

1088 self._nested_transaction is not None

1089 and self._nested_transaction.is_active

1090 )

1091

1092 def _is_autocommit_isolation(self) -> bool:

1093 opt_iso = self._execution_options.get("isolation_level", None)

1094 return bool(

1095 opt_iso == "AUTOCOMMIT"

1096 or (

1097 opt_iso is None

1098 and self.engine.dialect._on_connect_isolation_level

1099 == "AUTOCOMMIT"

1100 )

1101 )

1102

1103 def _get_required_transaction(self) -> RootTransaction:

1104 trans = self._transaction

1105 if trans is None:

1106 raise exc.InvalidRequestError("connection is not in a transaction")

1107 return trans

1108

1109 def _get_required_nested_transaction(self) -> NestedTransaction:

1110 trans = self._nested_transaction

1111 if trans is None:

1112 raise exc.InvalidRequestError(

1113 "connection is not in a nested transaction"

1114 )

1115 return trans

1116

1117 def get_transaction(self) -> Optional[RootTransaction]:

1118 """Return the current root transaction in progress, if any.

1119

1120 .. versionadded:: 1.4

1121

1122 """

1123

1124 return self._transaction

1125

1126 def get_nested_transaction(self) -> Optional[NestedTransaction]:

1127 """Return the current nested transaction in progress, if any.

1128

1129 .. versionadded:: 1.4

1130

1131 """

1132 return self._nested_transaction

1133

1134 def _begin_impl(self, transaction: RootTransaction) -> None:

1135 if self._echo:

1136 if self._is_autocommit_isolation():

1137 self._log_info(

1138 "BEGIN (implicit; DBAPI should not BEGIN due to "

1139 "autocommit mode)"

1140 )

1141 else:

1142 self._log_info("BEGIN (implicit)")

1143

1144 self.__in_begin = True

1145

1146 if self._has_events or self.engine._has_events:

1147 self.dispatch.begin(self)

1148

1149 try:

1150 self.engine.dialect.do_begin(self.connection)

1151 except BaseException as e:

1152 self._handle_dbapi_exception(e, None, None, None, None)

1153 finally:

1154 self.__in_begin = False

1155

1156 def _rollback_impl(self) -> None:

1157 if self._has_events or self.engine._has_events:

1158 self.dispatch.rollback(self)

1159

1160 if self._still_open_and_dbapi_connection_is_valid:

1161 if self._echo:

1162 if self._is_autocommit_isolation():

1163 if self.dialect.skip_autocommit_rollback:

1164 self._log_info(

1165 "ROLLBACK will be skipped by "

1166 "skip_autocommit_rollback"

1167 )

1168 else:

1169 self._log_info(

1170 "ROLLBACK using DBAPI connection.rollback(); "

1171 "set skip_autocommit_rollback to prevent fully"

1172 )

1173 else:

1174 self._log_info("ROLLBACK")

1175 try:

1176 self.engine.dialect.do_rollback(self.connection)

1177 except BaseException as e:

1178 self._handle_dbapi_exception(e, None, None, None, None)

1179

1180 def _commit_impl(self) -> None:

1181 if self._has_events or self.engine._has_events:

1182 self.dispatch.commit(self)

1183

1184 if self._echo:

1185 if self._is_autocommit_isolation():

1186 self._log_info(

1187 "COMMIT using DBAPI connection.commit(), "

1188 "has no effect due to autocommit mode"

1189 )

1190 else:

1191 self._log_info("COMMIT")

1192 try:

1193 self.engine.dialect.do_commit(self.connection)

1194 except BaseException as e:

1195 self._handle_dbapi_exception(e, None, None, None, None)

1196

1197 def _savepoint_impl(self, name: Optional[str] = None) -> str:

1198 if self._has_events or self.engine._has_events:

1199 self.dispatch.savepoint(self, name)

1200

1201 if name is None:

1202 self.__savepoint_seq += 1

1203 name = "sa_savepoint_%s" % self.__savepoint_seq

1204 self.engine.dialect.do_savepoint(self, name)

1205 return name

1206

1207 def _rollback_to_savepoint_impl(self, name: str) -> None:

1208 if self._has_events or self.engine._has_events:

1209 self.dispatch.rollback_savepoint(self, name, None)

1210

1211 if self._still_open_and_dbapi_connection_is_valid:

1212 self.engine.dialect.do_rollback_to_savepoint(self, name)

1213

1214 def _release_savepoint_impl(self, name: str) -> None:

1215 if self._has_events or self.engine._has_events:

1216 self.dispatch.release_savepoint(self, name, None)

1217

1218 self.engine.dialect.do_release_savepoint(self, name)

1219

1220 def _begin_twophase_impl(self, transaction: TwoPhaseTransaction) -> None:

1221 if self._echo:

1222 self._log_info("BEGIN TWOPHASE (implicit)")

1223 if self._has_events or self.engine._has_events:

1224 self.dispatch.begin_twophase(self, transaction.xid)

1225

1226 self.__in_begin = True

1227 try:

1228 self.engine.dialect.do_begin_twophase(self, transaction.xid)

1229 except BaseException as e:

1230 self._handle_dbapi_exception(e, None, None, None, None)

1231 finally:

1232 self.__in_begin = False

1233

1234 def _prepare_twophase_impl(self, xid: Any) -> None:

1235 if self._has_events or self.engine._has_events:

1236 self.dispatch.prepare_twophase(self, xid)

1237

1238 assert isinstance(self._transaction, TwoPhaseTransaction)

1239 try:

1240 self.engine.dialect.do_prepare_twophase(self, xid)

1241 except BaseException as e:

1242 self._handle_dbapi_exception(e, None, None, None, None)

1243

1244 def _rollback_twophase_impl(self, xid: Any, is_prepared: bool) -> None:

1245 if self._has_events or self.engine._has_events:

1246 self.dispatch.rollback_twophase(self, xid, is_prepared)

1247

1248 if self._still_open_and_dbapi_connection_is_valid:

1249 assert isinstance(self._transaction, TwoPhaseTransaction)

1250 try:

1251 self.engine.dialect.do_rollback_twophase(

1252 self, xid, is_prepared

1253 )

1254 except BaseException as e:

1255 self._handle_dbapi_exception(e, None, None, None, None)

1256

1257 def _commit_twophase_impl(self, xid: Any, is_prepared: bool) -> None:

1258 if self._has_events or self.engine._has_events:

1259 self.dispatch.commit_twophase(self, xid, is_prepared)

1260

1261 assert isinstance(self._transaction, TwoPhaseTransaction)

1262 try:

1263 self.engine.dialect.do_commit_twophase(self, xid, is_prepared)

1264 except BaseException as e:

1265 self._handle_dbapi_exception(e, None, None, None, None)

1266

1267 def close(self) -> None:

1268 """Close this :class:`_engine.Connection`.

1269

1270 This results in a release of the underlying database

1271 resources, that is, the DBAPI connection referenced

1272 internally. The DBAPI connection is typically restored

1273 back to the connection-holding :class:`_pool.Pool` referenced

1274 by the :class:`_engine.Engine` that produced this

1275 :class:`_engine.Connection`. Any transactional state present on

1276 the DBAPI connection is also unconditionally released via

1277 the DBAPI connection's ``rollback()`` method, regardless

1278 of any :class:`.Transaction` object that may be

1279 outstanding with regards to this :class:`_engine.Connection`.

1280

1281 This has the effect of also calling :meth:`_engine.Connection.rollback`

1282 if any transaction is in place.

1283

1284 After :meth:`_engine.Connection.close` is called, the

1285 :class:`_engine.Connection` is permanently in a closed state,

1286 and will allow no further operations.

1287

1288 """

1289

1290 # if the rollback here raises, skip_reset stays False and the pool

1291 # does its own reset of the connection, invalidating it if that

1292 # fails also. see #13657

1293 skip_reset = False

1294 try:

1295 if self._transaction:

1296 self._transaction.close()

1297 skip_reset = True

1298 finally:

1299 try:

1300 if self._dbapi_connection is not None:

1301 conn = self._dbapi_connection

1302

1303 # as we just closed the transaction, close the connection

1304 # pool connection without doing an additional reset

1305 if skip_reset:

1306 cast("_ConnectionFairy", conn)._close_special(

1307 transaction_reset=True

1308 )

1309 else:

1310 conn.close()

1311 finally:

1312 # There is a slight chance that conn.close() may have

1313 # triggered an invalidation here in which case

1314 # _dbapi_connection would already be None, however usually

1315 # it will be non-None here and in a "closed" state.

1316 #

1317 # there is also a chance that the conn.close() here could

1318 # itself be interrupted, such as by a cancellation; hence a

1319 # second try/finally is needed as well (see #13699)

1320 self._dbapi_connection = None

1321 self.__can_reconnect = False

1322

1323 # special case to handle mypy issue:

1324 # https://github.com/python/mypy/issues/20651

1325 @overload

1326 def scalar(

1327 self,

1328 statement: TypedReturnsRows[Never],

1329 parameters: Optional[_CoreSingleExecuteParams] = None,

1330 *,

1331 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1332 ) -> Optional[Any]: ...

1333

1334 @overload

1335 def scalar(

1336 self,

1337 statement: TypedReturnsRows[_T],

1338 parameters: Optional[_CoreSingleExecuteParams] = None,

1339 *,

1340 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1341 ) -> Optional[_T]: ...

1342

1343 @overload

1344 def scalar(

1345 self,

1346 statement: Executable,

1347 parameters: Optional[_CoreSingleExecuteParams] = None,

1348 *,

1349 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1350 ) -> Any: ...

1351

1352 def scalar(

1353 self,

1354 statement: Executable,

1355 parameters: Optional[_CoreSingleExecuteParams] = None,

1356 *,

1357 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1358 ) -> Any:

1359 r"""Executes a SQL statement construct and returns a scalar object.

1360

1361 This method is shorthand for invoking the

1362 :meth:`_engine.Result.scalar` method after invoking the

1363 :meth:`_engine.Connection.execute` method. Parameters are equivalent.

1364

1365 :return: a scalar Python value representing the first column of the

1366 first row returned.

1367

1368 """

1369 distilled_parameters = _distill_params_20(parameters)

1370 try:

1371 meth = statement._execute_on_scalar

1372 except AttributeError as err:

1373 raise exc.ObjectNotExecutableError(statement) from err

1374 else:

1375 return meth(

1376 self,

1377 distilled_parameters,

1378 execution_options or NO_OPTIONS,

1379 )

1380

1381 @overload

1382 def scalars(

1383 self,

1384 statement: TypedReturnsRows[_T],

1385 parameters: Optional[_CoreAnyExecuteParams] = None,

1386 *,

1387 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1388 ) -> ScalarResult[_T]: ...

1389

1390 @overload

1391 def scalars(

1392 self,

1393 statement: Executable,

1394 parameters: Optional[_CoreAnyExecuteParams] = None,

1395 *,

1396 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1397 ) -> ScalarResult[Any]: ...

1398

1399 def scalars(

1400 self,

1401 statement: Executable,

1402 parameters: Optional[_CoreAnyExecuteParams] = None,

1403 *,

1404 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1405 ) -> ScalarResult[Any]:

1406 """Executes and returns a scalar result set, which yields scalar values

1407 from the first column of each row.

1408

1409 This method is equivalent to calling :meth:`_engine.Connection.execute`

1410 to receive a :class:`_result.Result` object, then invoking the

1411 :meth:`_result.Result.scalars` method to produce a

1412 :class:`_result.ScalarResult` instance.

1413

1414 :return: a :class:`_result.ScalarResult`

1415

1416 .. versionadded:: 1.4.24

1417

1418 """

1419

1420 return self.execute(

1421 statement, parameters, execution_options=execution_options

1422 ).scalars()

1423

1424 @overload

1425 def execute(

1426 self,

1427 statement: TypedReturnsRows[Unpack[_Ts]],

1428 parameters: Optional[_CoreAnyExecuteParams] = None,

1429 *,

1430 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1431 ) -> CursorResult[Unpack[_Ts]]: ...

1432

1433 @overload

1434 def execute(

1435 self,

1436 statement: Executable,

1437 parameters: Optional[_CoreAnyExecuteParams] = None,

1438 *,

1439 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1440 ) -> CursorResult[Unpack[TupleAny]]: ...

1441

1442 def execute(

1443 self,

1444 statement: Executable,

1445 parameters: Optional[_CoreAnyExecuteParams] = None,

1446 *,

1447 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1448 ) -> CursorResult[Unpack[TupleAny]]:

1449 r"""Executes a SQL statement construct and returns a

1450 :class:`_engine.CursorResult`.

1451

1452 :param statement: The statement to be executed. This is always

1453 an object that is in both the :class:`_expression.ClauseElement` and

1454 :class:`_expression.Executable` hierarchies, including:

1455

1456 * :class:`_expression.Select`

1457 * :class:`_expression.Insert`, :class:`_expression.Update`,

1458 :class:`_expression.Delete`

1459 * :class:`_expression.TextClause` and

1460 :class:`_expression.TextualSelect`

1461 * :class:`_schema.DDL` and objects which inherit from

1462 :class:`_schema.ExecutableDDLElement`

1463

1464 :param parameters: parameters which will be bound into the statement.

1465 This may be either a dictionary of parameter names to values,

1466 or a mutable sequence (e.g. a list) of dictionaries. When a

1467 list of dictionaries is passed, the underlying statement execution

1468 will make use of the DBAPI ``cursor.executemany()`` method.

1469 When a single dictionary is passed, the DBAPI ``cursor.execute()``

1470 method will be used.

1471

1472 :param execution_options: optional dictionary of execution options,

1473 which will be associated with the statement execution. This

1474 dictionary can provide a subset of the options that are accepted

1475 by :meth:`_engine.Connection.execution_options`.

1476

1477 :return: a :class:`_engine.Result` object.

1478

1479 """

1480 distilled_parameters = _distill_params_20(parameters)

1481 try:

1482 meth = statement._execute_on_connection

1483 except AttributeError as err:

1484 raise exc.ObjectNotExecutableError(statement) from err

1485 else:

1486 return meth(

1487 self,

1488 distilled_parameters,

1489 execution_options or NO_OPTIONS,

1490 )

1491

1492 def _execute_function(

1493 self,

1494 func: FunctionElement[Any],

1495 distilled_parameters: _CoreMultiExecuteParams,

1496 execution_options: CoreExecuteOptionsParameter,

1497 ) -> CursorResult[Unpack[TupleAny]]:

1498 """Execute a sql.FunctionElement object."""

1499

1500 return self._execute_clauseelement(

1501 func.select(), distilled_parameters, execution_options

1502 )

1503

1504 def _execute_default(

1505 self,

1506 default: DefaultGenerator,

1507 distilled_parameters: _CoreMultiExecuteParams,

1508 execution_options: CoreExecuteOptionsParameter,

1509 ) -> Any:

1510 """Execute a schema.ColumnDefault object."""

1511

1512 exec_opts = self._execution_options.merge_with(execution_options)

1513

1514 event_multiparams: Optional[_CoreMultiExecuteParams]

1515 event_params: Optional[_CoreAnyExecuteParams]

1516

1517 # note for event handlers, the "distilled parameters" which is always

1518 # a list of dicts is broken out into separate "multiparams" and

1519 # "params" collections, which allows the handler to distinguish

1520 # between an executemany and execute style set of parameters.

1521 if self._has_events or self.engine._has_events:

1522 (

1523 default,

1524 distilled_parameters,

1525 event_multiparams,

1526 event_params,

1527 ) = self._invoke_before_exec_event(

1528 default, distilled_parameters, exec_opts

1529 )

1530 else:

1531 event_multiparams = event_params = None

1532

1533 try:

1534 conn = self._dbapi_connection

1535 if conn is None:

1536 conn = self._revalidate_connection()

1537

1538 dialect = self.dialect

1539 ctx = dialect.execution_ctx_cls._init_default(

1540 dialect, self, conn, exec_opts

1541 )

1542 except (exc.PendingRollbackError, exc.ResourceClosedError):

1543 raise

1544 except BaseException as e:

1545 self._handle_dbapi_exception(e, None, None, None, None)

1546

1547 ret = ctx._exec_default(None, default, None)

1548

1549 if self._has_events or self.engine._has_events:

1550 self.dispatch.after_execute(

1551 self,

1552 default,

1553 event_multiparams,

1554 event_params,

1555 exec_opts,

1556 ret,

1557 )

1558

1559 return ret

1560

1561 def _execute_ddl(

1562 self,

1563 ddl: ExecutableDDLElement,

1564 distilled_parameters: _CoreMultiExecuteParams,

1565 execution_options: CoreExecuteOptionsParameter,

1566 ) -> CursorResult[Unpack[TupleAny]]:

1567 """Execute a schema.DDL object."""

1568

1569 exec_opts = ddl._execution_options.merge_with(

1570 self._execution_options, execution_options

1571 )

1572

1573 event_multiparams: Optional[_CoreMultiExecuteParams]

1574 event_params: Optional[_CoreSingleExecuteParams]

1575

1576 if self._has_events or self.engine._has_events:

1577 (

1578 ddl,

1579 distilled_parameters,

1580 event_multiparams,

1581 event_params,

1582 ) = self._invoke_before_exec_event(

1583 ddl, distilled_parameters, exec_opts

1584 )

1585 else:

1586 event_multiparams = event_params = None

1587

1588 schema_translate_map = exec_opts.get("schema_translate_map", None)

1589

1590 dialect = self.dialect

1591

1592 compiled = ddl.compile(

1593 dialect=dialect, schema_translate_map=schema_translate_map

1594 )

1595 ret = self._execute_context(

1596 dialect,

1597 dialect.execution_ctx_cls._init_ddl,

1598 compiled,

1599 None,

1600 exec_opts,

1601 compiled,

1602 )

1603 if self._has_events or self.engine._has_events:

1604 self.dispatch.after_execute(

1605 self,

1606 ddl,

1607 event_multiparams,

1608 event_params,

1609 exec_opts,

1610 ret,

1611 )

1612 return ret

1613

1614 def _invoke_before_exec_event(

1615 self,

1616 elem: Any,

1617 distilled_params: _CoreMultiExecuteParams,

1618 execution_options: _ExecuteOptions,

1619 ) -> Tuple[

1620 Any,

1621 _CoreMultiExecuteParams,

1622 _CoreMultiExecuteParams,

1623 _CoreSingleExecuteParams,

1624 ]:

1625 event_multiparams: _CoreMultiExecuteParams

1626 event_params: _CoreSingleExecuteParams

1627

1628 if len(distilled_params) == 1:

1629 event_multiparams, event_params = [], distilled_params[0]

1630 else:

1631 event_multiparams, event_params = distilled_params, {}

1632

1633 for fn in self.dispatch.before_execute:

1634 elem, event_multiparams, event_params = fn(

1635 self,

1636 elem,

1637 event_multiparams,

1638 event_params,

1639 execution_options,

1640 )

1641

1642 if event_multiparams:

1643 distilled_params = list(event_multiparams)

1644 if event_params:

1645 raise exc.InvalidRequestError(

1646 "Event handler can't return non-empty multiparams "

1647 "and params at the same time"

1648 )

1649 elif event_params:

1650 distilled_params = [event_params]

1651 else:

1652 distilled_params = []

1653

1654 return elem, distilled_params, event_multiparams, event_params

1655

1656 def _execute_clauseelement(

1657 self,

1658 elem: Executable,

1659 distilled_parameters: _CoreMultiExecuteParams,

1660 execution_options: CoreExecuteOptionsParameter,

1661 ) -> CursorResult[Unpack[TupleAny]]:

1662 """Execute a sql.ClauseElement object."""

1663

1664 exec_opts = elem._execution_options.merge_with(

1665 self._execution_options, execution_options

1666 )

1667

1668 has_events = self._has_events or self.engine._has_events

1669 if has_events:

1670 (

1671 elem,

1672 distilled_parameters,

1673 event_multiparams,

1674 event_params,

1675 ) = self._invoke_before_exec_event(

1676 elem, distilled_parameters, exec_opts

1677 )

1678

1679 if distilled_parameters:

1680 # ensure we don't retain a link to the view object for keys()

1681 # which links to the values, which we don't want to cache

1682 keys = sorted(distilled_parameters[0])

1683 for_executemany = len(distilled_parameters) > 1

1684 else:

1685 keys = []

1686 for_executemany = False

1687

1688 dialect = self.dialect

1689

1690 schema_translate_map = exec_opts.get("schema_translate_map", None)

1691

1692 compiled_cache: Optional[CompiledCacheType] = exec_opts.get(

1693 "compiled_cache", self.engine._compiled_cache

1694 )

1695

1696 compiled_sql, extracted_params, param_dict, cache_hit = (

1697 elem._compile_w_cache(

1698 dialect=dialect,

1699 compiled_cache=compiled_cache,

1700 column_keys=keys,

1701 for_executemany=for_executemany,

1702 schema_translate_map=schema_translate_map,

1703 linting=self.dialect.compiler_linting | compiler.WARN_LINTING,

1704 )

1705 )

1706 ret = self._execute_context(

1707 dialect,

1708 dialect.execution_ctx_cls._init_compiled,

1709 compiled_sql,

1710 distilled_parameters,

1711 exec_opts,

1712 compiled_sql,

1713 distilled_parameters,

1714 elem,

1715 extracted_params,

1716 cache_hit=cache_hit,

1717 param_dict=param_dict,

1718 )

1719 if has_events:

1720 self.dispatch.after_execute(

1721 self,

1722 elem,

1723 event_multiparams,

1724 event_params,

1725 exec_opts,

1726 ret,

1727 )

1728 return ret

1729

1730 def exec_driver_sql(

1731 self,

1732 statement: str,

1733 parameters: Optional[_DBAPIAnyExecuteParams] = None,

1734 execution_options: Optional[CoreExecuteOptionsParameter] = None,

1735 ) -> CursorResult[Unpack[TupleAny]]:

1736 r"""Executes a string SQL statement on the DBAPI cursor directly,

1737 without any SQL compilation steps.

1738

1739 This can be used to pass any string directly to the

1740 ``cursor.execute()`` method of the DBAPI in use.

1741

1742 :param statement: The statement str to be executed. Bound parameters

1743 must use the underlying DBAPI's paramstyle, such as "qmark",

1744 "pyformat", "format", etc.

1745

1746 :param parameters: represent bound parameter values to be used in the

1747 execution. The format is one of: a dictionary of named parameters,

1748 a tuple of positional parameters, or a list containing either

1749 dictionaries or tuples for multiple-execute support.

1750

1751 :return: a :class:`_engine.CursorResult`.

1752

1753 E.g. multiple dictionaries::

1754

1755

1756 conn.exec_driver_sql(

1757 "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)",

1758 [{"id": 1, "value": "v1"}, {"id": 2, "value": "v2"}],

1759 )

1760

1761 Single dictionary::

1762

1763 conn.exec_driver_sql(

1764 "INSERT INTO table (id, value) VALUES (%(id)s, %(value)s)",

1765 dict(id=1, value="v1"),

1766 )

1767

1768 Single tuple::

1769

1770 conn.exec_driver_sql(

1771 "INSERT INTO table (id, value) VALUES (?, ?)", (1, "v1")

1772 )

1773

1774 .. note:: The :meth:`_engine.Connection.exec_driver_sql` method does

1775 not participate in the

1776 :meth:`_events.ConnectionEvents.before_execute` and

1777 :meth:`_events.ConnectionEvents.after_execute` events. To

1778 intercept calls to :meth:`_engine.Connection.exec_driver_sql`, use

1779 :meth:`_events.ConnectionEvents.before_cursor_execute` and

1780 :meth:`_events.ConnectionEvents.after_cursor_execute`.

1782 .. seealso::

1784 :pep:`249`

1786 """

1788 distilled_parameters = _distill_raw_params(parameters)

1789

1790 exec_opts = self._execution_options.merge_with(execution_options)

1791

1792 dialect = self.dialect

1793 ret = self._execute_context(

1794 dialect,

1795 dialect.execution_ctx_cls._init_statement,

1796 statement,

1797 None,

1798 exec_opts,

1799 statement,

1800 distilled_parameters,

1801 )

1802

1803 return ret

1804

1805 def _execute_context(

1806 self,

1807 dialect: Dialect,

1808 constructor: Callable[..., ExecutionContext],

1809 statement: Union[str, Compiled],

1810 parameters: Optional[_AnyMultiExecuteParams],

1811 execution_options: _ExecuteOptions,

1812 *args: Any,

1813 **kw: Any,

1814 ) -> CursorResult[Unpack[TupleAny]]:

1815 """Create an :class:`.ExecutionContext` and execute, returning

1816 a :class:`_engine.CursorResult`."""

1817

1818 if execution_options:

1819 yp = execution_options.get("yield_per", None)

1820 if yp:

1821 execution_options = execution_options.union(

1822 {"stream_results": True, "max_row_buffer": yp}

1823 )

1824 try:

1825 conn = self._dbapi_connection

1826 if conn is None:

1827 conn = self._revalidate_connection()

1828

1829 context = constructor(

1830 dialect, self, conn, execution_options, *args, **kw

1831 )

1832 except (exc.PendingRollbackError, exc.ResourceClosedError):

1833 raise

1834 except BaseException as e:

1835 self._handle_dbapi_exception(

1836 e, str(statement), parameters, None, None

1837 )

1838

1839 if (

1840 self._transaction

1841 and not self._transaction.is_active

1842 or (

1843 self._nested_transaction

1844 and not self._nested_transaction.is_active

1845 )

1846 ):

1847 self._invalid_transaction()

1848

1849 elif self._trans_context_manager:

1850 TransactionalContext._trans_ctx_check(self)

1851

1852 if self._transaction is None:

1853 self._autobegin()

1854

1855 context.pre_exec()

1856

1857 if context.execute_style is ExecuteStyle.INSERTMANYVALUES:

1858 return self._exec_insertmany_context(dialect, context)

1859 else:

1860 return self._exec_single_context(

1861 dialect, context, statement, parameters

1862 )

1863

1864 def _exec_single_context(

1865 self,

1866 dialect: Dialect,

1867 context: ExecutionContext,

1868 statement: Union[str, Compiled],

1869 parameters: Optional[_AnyMultiExecuteParams],

1870 ) -> CursorResult[Unpack[TupleAny]]:

1871 """continue the _execute_context() method for a single DBAPI

1872 cursor.execute() or cursor.executemany() call.

1873

1874 """

1875 if dialect.bind_typing is BindTyping.SETINPUTSIZES:

1876 generic_setinputsizes = context._prepare_set_input_sizes()

1877

1878 if generic_setinputsizes:

1879 try:

1880 dialect.do_set_input_sizes(

1881 context.cursor, generic_setinputsizes, context

1882 )

1883 except BaseException as e:

1884 self._handle_dbapi_exception(

1885 e, str(statement), parameters, None, context

1886 )

1887

1888 cursor, str_statement, parameters = (

1889 context.cursor,

1890 context.statement,

1891 context.parameters,

1892 )

1893

1894 effective_parameters: Optional[_AnyExecuteParams]

1895

1896 if not context.executemany:

1897 effective_parameters = parameters[0]

1898 else:

1899 effective_parameters = parameters

1900

1901 evt_handled: bool = False

1902 try:

1903 if self._has_events or self.engine._has_events:

1904 for fn in self.dispatch.before_cursor_execute:

1905 str_statement, effective_parameters = fn(

1906 self,

1907 cursor,

1908 str_statement,

1909 effective_parameters,

1910 context,

1911 context.executemany,

1912 )

1913

1914 if self._echo:

1915 self._log_info(str_statement)

1916

1917 stats = context._get_cache_stats()

1918

1919 self._log_parameters(

1920 effective_parameters, stats, ismulti=context.executemany

1921 )

1922

1923 if context.execute_style is ExecuteStyle.EXECUTEMANY:

1924 effective_parameters = cast(

1925 "_CoreMultiExecuteParams", effective_parameters

1926 )

1927 if self.dialect._has_events:

1928 for fn in self.dialect.dispatch.do_executemany:

1929 if fn(

1930 cursor,

1931 str_statement,

1932 effective_parameters,

1933 context,

1934 ):

1935 evt_handled = True

1936 break

1937 if not evt_handled:

1938 self.dialect.do_executemany(

1939 cursor,

1940 str_statement,

1941 effective_parameters,

1942 context,

1943 )

1944 elif not effective_parameters and context.no_parameters:

1945 if self.dialect._has_events:

1946 for fn in self.dialect.dispatch.do_execute_no_params:

1947 if fn(cursor, str_statement, context):

1948 evt_handled = True

1949 break

1950 if not evt_handled:

1951 self.dialect.do_execute_no_params(

1952 cursor, str_statement, context

1953 )

1954 else:

1955 effective_parameters = cast(

1956 "_CoreSingleExecuteParams", effective_parameters

1957 )

1958 if self.dialect._has_events:

1959 for fn in self.dialect.dispatch.do_execute:

1960 if fn(

1961 cursor,

1962 str_statement,

1963 effective_parameters,

1964 context,

1965 ):

1966 evt_handled = True

1967 break

1968 if not evt_handled:

1969 self.dialect.do_execute(

1970 cursor, str_statement, effective_parameters, context

1971 )

1972

1973 if self._has_events or self.engine._has_events:

1974 self.dispatch.after_cursor_execute(

1975 self,

1976 cursor,

1977 str_statement,

1978 effective_parameters,

1979 context,

1980 context.executemany,

1981 )

1982

1983 context.post_exec()

1984

1985 result = context._setup_result_proxy()

1986

1987 except BaseException as e:

1988 self._handle_dbapi_exception(

1989 e, str_statement, effective_parameters, cursor, context

1990 )

1991

1992 return result

1993

1994 def _exec_insertmany_context(

1995 self,

1996 dialect: Dialect,

1997 context: ExecutionContext,

1998 ) -> CursorResult[Unpack[TupleAny]]:

1999 """continue the _execute_context() method for an "insertmanyvalues"

2000 operation, which will invoke DBAPI

2001 cursor.execute() one or more times with individual log and

2002 event hook calls.

2003

2004 """

2005

2006 if dialect.bind_typing is BindTyping.SETINPUTSIZES:

2007 generic_setinputsizes = context._prepare_set_input_sizes()

2008 else:

2009 generic_setinputsizes = None

2010

2011 cursor, str_statement, parameters = (

2012 context.cursor,

2013 context.statement,

2014 context.parameters,

2015 )

2016

2017 effective_parameters = parameters

2018

2019 engine_events = self._has_events or self.engine._has_events

2020 if self.dialect._has_events:

2021 do_execute_dispatch: Iterable[Any] = (

2022 self.dialect.dispatch.do_execute

2023 )

2024 else:

2025 do_execute_dispatch = ()

2026

2027 if self._echo:

2028 stats = context._get_cache_stats() + " (insertmanyvalues)"

2029

2030 preserve_rowcount = context.execution_options.get(

2031 "preserve_rowcount", False

2032 )

2033 rowcount = 0

2034

2035 for imv_batch in dialect._deliver_insertmanyvalues_batches(

2036 self,

2037 cursor,

2038 str_statement,

2039 effective_parameters,

2040 generic_setinputsizes,

2041 context,

2042 ):

2043 if imv_batch.processed_setinputsizes:

2044 try:

2045 dialect.do_set_input_sizes(

2046 context.cursor,

2047 imv_batch.processed_setinputsizes,

2048 context,

2049 )

2050 except BaseException as e:

2051 self._handle_dbapi_exception(

2052 e,

2053 sql_util._long_statement(imv_batch.replaced_statement),

2054 imv_batch.replaced_parameters,

2055 None,

2056 context,

2057 is_sub_exec=True,

2058 )

2059

2060 sub_stmt = imv_batch.replaced_statement

2061 sub_params = imv_batch.replaced_parameters

2062

2063 try:

2064 if engine_events:

2065 for fn in self.dispatch.before_cursor_execute:

2066 sub_stmt, sub_params = fn(

2067 self,

2068 cursor,

2069 sub_stmt,

2070 sub_params,

2071 context,

2072 True,

2073 )

2074

2075 if self._echo:

2076 self._log_info(sql_util._long_statement(sub_stmt))

2077

2078 imv_stats = f""" {imv_batch.batchnum}/{

2079 imv_batch.total_batches

2080 } ({

2081 'ordered'

2082 if imv_batch.rows_sorted else 'unordered'

2083 }{

2084 '; batch not supported'

2085 if imv_batch.is_downgraded

2086 else ''

2087 })"""

2088

2089 if imv_batch.batchnum == 1:

2090 stats += imv_stats

2091 else:

2092 stats = f"insertmanyvalues{imv_stats}"

2093

2094 self._log_parameters(sub_params, stats, ismulti=False)

2095

2096 for fn in do_execute_dispatch:

2097 if fn(

2098 cursor,

2099 sub_stmt,

2100 sub_params,

2101 context,

2102 ):

2103 break

2104 else:

2105 dialect.do_execute(

2106 cursor,

2107 sub_stmt,

2108 sub_params,

2109 context,

2110 )

2111

2112 if engine_events:

2113 self.dispatch.after_cursor_execute(

2114 self,

2115 cursor,

2116 sub_stmt,

2117 sub_params,

2118 context,

2119 context.executemany,

2120 )

2121 except BaseException as e:

2122 self._handle_dbapi_exception(

2123 e,

2124 sql_util._long_statement(sub_stmt),

2125 sub_params,

2126 cursor,

2127 context,

2128 is_sub_exec=True,

2129 )

2130

2131 if preserve_rowcount:

2132 rowcount += imv_batch.current_batch_size

2133

2134 try:

2135 context.post_exec()

2136

2137 if preserve_rowcount:

2138 context._rowcount = rowcount # type: ignore[attr-defined]

2139

2140 result = context._setup_result_proxy()

2141

2142 except BaseException as e:

2143 self._handle_dbapi_exception(

2144 e, str_statement, effective_parameters, cursor, context

2145 )

2146

2147 return result

2148

2149 def _cursor_execute(

2150 self,

2151 cursor: DBAPICursor,

2152 statement: str,

2153 parameters: _DBAPISingleExecuteParams,

2154 context: Optional[ExecutionContext] = None,

2155 ) -> None:

2156 """Execute a statement + params on the given cursor.

2157

2158 Adds appropriate logging and exception handling.

2159

2160 This method is used by DefaultDialect for special-case

2161 executions, such as for sequences and column defaults.

2162 The path of statement execution in the majority of cases

2163 terminates at _execute_context().

2164

2165 """

2166 try:

2167 if self._has_events or self.engine._has_events:

2168 for fn in self.dispatch.before_cursor_execute:

2169 statement, parameters = fn(

2170 self, cursor, statement, parameters, context, False

2171 )

2172

2173 if self._echo:

2174 self._log_info(statement)

2175 self._log_parameters(parameters, "raw sql", ismulti=False)

2176

2177 for fn in (

2178 ()

2179 if not self.dialect._has_events

2180 else self.dialect.dispatch.do_execute

2181 ):

2182 if fn(cursor, statement, parameters, context):

2183 break

2184 else:

2185 self.dialect.do_execute(cursor, statement, parameters, context)

2186

2187 if self._has_events or self.engine._has_events:

2188 self.dispatch.after_cursor_execute(

2189 self, cursor, statement, parameters, context, False

2190 )

2191 except BaseException as e:

2192 self._handle_dbapi_exception(

2193 e, statement, parameters, cursor, context

2194 )

2195

2196 def _safe_close_cursor(self, cursor: DBAPICursor) -> None:

2197 """Close the given cursor, catching exceptions

2198 and turning into log warnings.

2199

2200 """

2201 try:

2202 cursor.close()

2203 except Exception:

2204 # log the error through the connection pool's logger.

2205 self.engine.pool.logger.error(

2206 "Error closing cursor", exc_info=True

2207 )

2208

2209 _reentrant_error = False

2210 _is_disconnect = False

2211

2212 def _handle_dbapi_exception(

2213 self,

2214 e: BaseException,

2215 statement: Optional[str],

2216 parameters: Optional[_AnyExecuteParams],

2217 cursor: Optional[DBAPICursor],

2218 context: Optional[ExecutionContext],

2219 is_sub_exec: bool = False,

2220 ) -> NoReturn:

2221 exc_info = sys.exc_info()

2222

2223 is_exit_exception = util.is_exit_exception(e)

2224

2225 if not self._is_disconnect:

2226 self._is_disconnect = (

2227 isinstance(e, self.dialect.loaded_dbapi.Error)

2228 and not self.closed

2229 and self.dialect.is_disconnect(

2230 e,

2231 self._dbapi_connection if not self.invalidated else None,

2232 cursor,

2233 )

2234 ) or (is_exit_exception and not self.closed)

2235

2236 invalidate_pool_on_disconnect = not is_exit_exception

2237

2238 ismulti: bool = (

2239 not is_sub_exec and context.executemany

2240 if context is not None

2241 else False

2242 )

2243 if self._reentrant_error:

2244 raise exc.DBAPIError.instance(

2245 statement,

2246 parameters,

2247 e,

2248 self.dialect.loaded_dbapi.Error,

2249 hide_parameters=self.engine.hide_parameters,

2250 dialect=self.dialect,

2251 ismulti=ismulti,

2252 ).with_traceback(exc_info[2]) from e

2253 self._reentrant_error = True

2254 try:

2255 # non-DBAPI error - if we already got a context,

2256 # or there's no string statement, don't wrap it

2257 should_wrap = isinstance(e, self.dialect.loaded_dbapi.Error) or (

2258 not isinstance(e, exc.StatementError)

2259 and statement is not None

2260 and context is None

2261 and not is_exit_exception

2262 )

2263

2264 if should_wrap:

2265 sqlalchemy_exception = exc.DBAPIError.instance(

2266 statement,

2267 parameters,

2268 cast(Exception, e),

2269 self.dialect.loaded_dbapi.Error,

2270 hide_parameters=self.engine.hide_parameters,

2271 connection_invalidated=self._is_disconnect,

2272 dialect=self.dialect,

2273 ismulti=ismulti,

2274 )

2275 else:

2276 sqlalchemy_exception = None

2277

2278 newraise = None

2279

2280 if (self.dialect._has_events) and not self._execution_options.get(

2281 "skip_user_error_events", False

2282 ):

2283 ctx = ExceptionContextImpl(

2284 e,

2285 sqlalchemy_exception,

2286 self.engine,

2287 self.dialect,

2288 self,

2289 cursor,

2290 statement,

2291 parameters,

2292 context,

2293 self._is_disconnect,

2294 invalidate_pool_on_disconnect,

2295 False,

2296 )

2297

2298 for fn in self.dialect.dispatch.handle_error:

2299 try:

2300 # handler returns an exception;

2301 # call next handler in a chain

2302 per_fn = fn(ctx)

2303 if per_fn is not None:

2304 ctx.chained_exception = newraise = per_fn

2305 except Exception as _raised:

2306 # handler raises an exception - stop processing

2307 newraise = _raised

2308 break

2309

2310 if self._is_disconnect != ctx.is_disconnect:

2311 self._is_disconnect = ctx.is_disconnect

2312 if sqlalchemy_exception:

2313 sqlalchemy_exception.connection_invalidated = (

2314 ctx.is_disconnect

2315 )

2316

2317 # set up potentially user-defined value for

2318 # invalidate pool.

2319 invalidate_pool_on_disconnect = (

2320 ctx.invalidate_pool_on_disconnect

2321 )

2322

2323 if should_wrap and context:

2324 context.handle_dbapi_exception(e)

2325

2326 if not self._is_disconnect:

2327 if cursor:

2328 self._safe_close_cursor(cursor)

2329 # "autorollback" was mostly relevant in 1.x series.

2330 # It's very unlikely to reach here, as the connection

2331 # does autobegin so when we are here, we are usually

2332 # in an explicit / semi-explicit transaction.

2333 # however we have a test which manufactures this

2334 # scenario in any case using an event handler.

2335 # test/engine/test_execute.py-> test_actual_autorollback

2336 if not self.in_transaction():

2337 self._rollback_impl()

2338

2339 if newraise:

2340 raise newraise.with_traceback(exc_info[2]) from e

2341 elif should_wrap:

2342 assert sqlalchemy_exception is not None

2343 raise sqlalchemy_exception.with_traceback(exc_info[2]) from e

2344 else:

2345 assert exc_info[1] is not None

2346 raise exc_info[1].with_traceback(exc_info[2])

2347 finally:

2348 del self._reentrant_error

2349 if self._is_disconnect:

2350 del self._is_disconnect

2351 if not self.invalidated:

2352 dbapi_conn_wrapper = self._dbapi_connection

2353 assert dbapi_conn_wrapper is not None

2354 if invalidate_pool_on_disconnect:

2355 self.engine.pool._invalidate(dbapi_conn_wrapper, e)

2356 self.invalidate(e)

2357

2358 @classmethod

2359 def _handle_dbapi_exception_noconnection(

2360 cls,

2361 e: BaseException,

2362 dialect: Dialect,

2363 engine: Optional[Engine] = None,

2364 is_disconnect: Optional[bool] = None,

2365 invalidate_pool_on_disconnect: bool = True,

2366 is_pre_ping: bool = False,

2367 ) -> NoReturn:

2368 exc_info = sys.exc_info()

2369

2370 if is_disconnect is None:

2371 is_disconnect = isinstance(

2372 e, dialect.loaded_dbapi.Error

2373 ) and dialect.is_disconnect(e, None, None)

2374

2375 should_wrap = isinstance(e, dialect.loaded_dbapi.Error)

2376

2377 if should_wrap:

2378 sqlalchemy_exception = exc.DBAPIError.instance(

2379 None,

2380 None,

2381 cast(Exception, e),

2382 dialect.loaded_dbapi.Error,

2383 hide_parameters=(

2384 engine.hide_parameters if engine is not None else False

2385 ),

2386 connection_invalidated=is_disconnect,

2387 dialect=dialect,

2388 )

2389 else:

2390 sqlalchemy_exception = None

2391

2392 newraise = None

2393

2394 if dialect._has_events:

2395 ctx = ExceptionContextImpl(

2396 e,

2397 sqlalchemy_exception,

2398 engine,

2399 dialect,

2400 None,

2401 None,

2402 None,

2403 None,

2404 None,

2405 is_disconnect,

2406 invalidate_pool_on_disconnect,

2407 is_pre_ping,

2408 )

2409 for fn in dialect.dispatch.handle_error:

2410 try:

2411 # handler returns an exception;

2412 # call next handler in a chain

2413 per_fn = fn(ctx)

2414 if per_fn is not None:

2415 ctx.chained_exception = newraise = per_fn

2416 except Exception as _raised:

2417 # handler raises an exception - stop processing

2418 newraise = _raised

2419 break

2420

2421 if sqlalchemy_exception and is_disconnect != ctx.is_disconnect:

2422 sqlalchemy_exception.connection_invalidated = ctx.is_disconnect

2423

2424 if newraise:

2425 raise newraise.with_traceback(exc_info[2]) from e

2426 elif should_wrap:

2427 assert sqlalchemy_exception is not None

2428 raise sqlalchemy_exception.with_traceback(exc_info[2]) from e

2429 else:

2430 assert exc_info[1] is not None

2431 raise exc_info[1].with_traceback(exc_info[2])

2432

2433 def _run_ddl_visitor(

2434 self,

2435 visitorcallable: Type[InvokeDDLBase],

2436 element: SchemaVisitable,

2437 **kwargs: Any,

2438 ) -> None:

2439 """run a DDL visitor.

2440

2441 This method is only here so that the MockConnection can change the

2442 options given to the visitor so that "checkfirst" is skipped.

2443

2444 """

2445 visitorcallable(

2446 dialect=self.dialect, connection=self, **kwargs

2447 ).traverse_single(element)

2448

2449

2450class ExceptionContextImpl(ExceptionContext):

2451 """Implement the :class:`.ExceptionContext` interface."""

2452

2453 __slots__ = (

2454 "connection",

2455 "engine",

2456 "dialect",

2457 "cursor",

2458 "statement",

2459 "parameters",

2460 "original_exception",

2461 "sqlalchemy_exception",

2462 "chained_exception",

2463 "execution_context",

2464 "is_disconnect",

2465 "invalidate_pool_on_disconnect",

2466 "is_pre_ping",

2467 )

2468

2469 def __init__(

2470 self,

2471 exception: BaseException,

2472 sqlalchemy_exception: Optional[exc.StatementError],

2473 engine: Optional[Engine],

2474 dialect: Dialect,

2475 connection: Optional[Connection],

2476 cursor: Optional[DBAPICursor],

2477 statement: Optional[str],

2478 parameters: Optional[_DBAPIAnyExecuteParams],

2479 context: Optional[ExecutionContext],

2480 is_disconnect: bool,

2481 invalidate_pool_on_disconnect: bool,

2482 is_pre_ping: bool,

2483 ):

2484 self.engine = engine

2485 self.dialect = dialect

2486 self.connection = connection

2487 self.sqlalchemy_exception = sqlalchemy_exception

2488 self.original_exception = exception

2489 self.execution_context = context

2490 self.statement = statement

2491 self.parameters = parameters

2492 self.is_disconnect = is_disconnect

2493 self.invalidate_pool_on_disconnect = invalidate_pool_on_disconnect

2494 self.is_pre_ping = is_pre_ping

2495

2496

2497class Transaction(TransactionalContext):

2498 """Represent a database transaction in progress.

2499

2500 The :class:`.Transaction` object is procured by

2501 calling the :meth:`_engine.Connection.begin` method of

2502 :class:`_engine.Connection`::

2503

2504 from sqlalchemy import create_engine

2505

2506 engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")

2507 connection = engine.connect()

2508 trans = connection.begin()

2509 connection.execute(text("insert into x (a, b) values (1, 2)"))

2510 trans.commit()

2511

2512 The object provides :meth:`.rollback` and :meth:`.commit`

2513 methods in order to control transaction boundaries. It

2514 also implements a context manager interface so that

2515 the Python ``with`` statement can be used with the

2516 :meth:`_engine.Connection.begin` method::

2517

2518 with connection.begin():

2519 connection.execute(text("insert into x (a, b) values (1, 2)"))

2520

2521 The Transaction object is **not** threadsafe.

2522

2523 .. seealso::

2524

2525 :meth:`_engine.Connection.begin`

2526

2527 :meth:`_engine.Connection.begin_twophase`

2528

2529 :meth:`_engine.Connection.begin_nested`

2530

2531 .. index::

2532 single: thread safety; Transaction

2533 """ # noqa

2534

2535 __slots__ = ()

2536

2537 _is_root: bool = False

2538 is_active: bool

2539 connection: Connection

2540

2541 def __init__(self, connection: Connection):

2542 raise NotImplementedError()

2543

2544 @property

2545 def _deactivated_from_connection(self) -> bool:

2546 """True if this transaction is totally deactivated from the connection

2547 and therefore can no longer affect its state.

2548

2549 """

2550 raise NotImplementedError()

2551

2552 def _do_close(self) -> None:

2553 raise NotImplementedError()

2554

2555 def _do_rollback(self) -> None:

2556 raise NotImplementedError()

2557

2558 def _do_commit(self) -> None:

2559 raise NotImplementedError()

2560

2561 @property

2562 def is_valid(self) -> bool:

2563 return self.is_active and not self.connection.invalidated

2564

2565 def close(self) -> None:

2566 """Close this :class:`.Transaction`.

2567

2568 If this transaction is the base transaction in a begin/commit

2569 nesting, the transaction will rollback(). Otherwise, the

2570 method returns.

2571

2572 This is used to cancel a Transaction without affecting the scope of

2573 an enclosing transaction.

2574

2575 """

2576 try:

2577 self._do_close()

2578 finally:

2579 assert not self.is_active

2580

2581 def rollback(self) -> None:

2582 """Roll back this :class:`.Transaction`.

2583

2584 The implementation of this may vary based on the type of transaction in

2585 use:

2586

2587 * For a simple database transaction (e.g. :class:`.RootTransaction`),

2588 it corresponds to a ROLLBACK.

2589

2590 * For a :class:`.NestedTransaction`, it corresponds to a

2591 "ROLLBACK TO SAVEPOINT" operation.

2592

2593 * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two

2594 phase transactions may be used.

2595

2596

2597 """

2598 try:

2599 self._do_rollback()

2600 finally:

2601 assert not self.is_active

2602

2603 def commit(self) -> None:

2604 """Commit this :class:`.Transaction`.

2605

2606 The implementation of this may vary based on the type of transaction in

2607 use:

2608

2609 * For a simple database transaction (e.g. :class:`.RootTransaction`),

2610 it corresponds to a COMMIT.

2611

2612 * For a :class:`.NestedTransaction`, it corresponds to a

2613 "RELEASE SAVEPOINT" operation.

2614

2615 * For a :class:`.TwoPhaseTransaction`, DBAPI-specific methods for two

2616 phase transactions may be used.

2617

2618 """

2619 try:

2620 self._do_commit()

2621 finally:

2622 assert not self.is_active

2623

2624 def _get_subject(self) -> Connection:

2625 return self.connection

2626

2627 def _transaction_is_active(self) -> bool:

2628 return self.is_active

2629

2630 def _transaction_is_closed(self) -> bool:

2631 return not self._deactivated_from_connection

2632

2633 def _rollback_can_be_called(self) -> bool:

2634 # for RootTransaction / NestedTransaction, it's safe to call

2635 # rollback() even if the transaction is deactive and no warnings

2636 # will be emitted. tested in

2637 # test_transaction.py -> test_no_rollback_in_deactive(?:_savepoint)?

2638 return True

2639

2640

2641class RootTransaction(Transaction):

2642 """Represent the "root" transaction on a :class:`_engine.Connection`.

2643

2644 This corresponds to the current "BEGIN/COMMIT/ROLLBACK" that's occurring

2645 for the :class:`_engine.Connection`. The :class:`_engine.RootTransaction`

2646 is created by calling upon the :meth:`_engine.Connection.begin` method, and

2647 remains associated with the :class:`_engine.Connection` throughout its

2648 active span. The current :class:`_engine.RootTransaction` in use is

2649 accessible via the :attr:`_engine.Connection.get_transaction` method of

2650 :class:`_engine.Connection`.

2651

2652 In :term:`2.0 style` use, the :class:`_engine.Connection` also employs

2653 "autobegin" behavior that will create a new

2654 :class:`_engine.RootTransaction` whenever a connection in a

2655 non-transactional state is used to emit commands on the DBAPI connection.

2656 The scope of the :class:`_engine.RootTransaction` in 2.0 style

2657 use can be controlled using the :meth:`_engine.Connection.commit` and

2658 :meth:`_engine.Connection.rollback` methods.

2661 """

2663 _is_root = True

2665 __slots__ = ("connection", "is_active")

2666

2667 def __init__(self, connection: Connection):

2668 assert connection._transaction is None

2669 if connection._trans_context_manager:

2670 TransactionalContext._trans_ctx_check(connection)

2671 self.connection = connection

2672 self._connection_begin_impl()

2673 connection._transaction = self

2674

2675 self.is_active = True

2676

2677 def _deactivate_from_connection(self) -> None:

2678 if self.is_active:

2679 assert self.connection._transaction is self

2680 self.is_active = False

2681

2682 elif self.connection._transaction is not self:

2683 util.warn("transaction already deassociated from connection")

2684

2685 @property

2686 def _deactivated_from_connection(self) -> bool:

2687 return self.connection._transaction is not self

2688

2689 def _connection_begin_impl(self) -> None:

2690 self.connection._begin_impl(self)

2691

2692 def _connection_rollback_impl(self) -> None:

2693 self.connection._rollback_impl()

2694

2695 def _connection_commit_impl(self) -> None:

2696 self.connection._commit_impl()

2697

2698 def _close_impl(self, try_deactivate: bool = False) -> None:

2699 try:

2700 if self.is_active:

2701 self._connection_rollback_impl()

2702

2703 if self.connection._nested_transaction:

2704 self.connection._nested_transaction._cancel()

2705 finally:

2706 if self.is_active or try_deactivate:

2707 self._deactivate_from_connection()

2708 if self.connection._transaction is self:

2709 self.connection._transaction = None

2710

2711 assert not self.is_active

2712 assert self.connection._transaction is not self

2713

2714 def _do_close(self) -> None:

2715 self._close_impl()

2716

2717 def _do_rollback(self) -> None:

2718 self._close_impl(try_deactivate=True)

2719

2720 def _do_commit(self) -> None:

2721 if self.is_active:

2722 assert self.connection._transaction is self

2723

2724 try:

2725 self._connection_commit_impl()

2726 finally:

2727 # whether or not commit succeeds, cancel any

2728 # nested transactions, make this transaction "inactive"

2729 # and remove it as a reset agent

2730 if self.connection._nested_transaction:

2731 self.connection._nested_transaction._cancel()

2732

2733 self._deactivate_from_connection()

2734

2735 # ...however only remove as the connection's current transaction

2736 # if commit succeeded. otherwise it stays on so that a rollback

2737 # needs to occur.

2738 self.connection._transaction = None

2739 else:

2740 if self.connection._transaction is self:

2741 self.connection._invalid_transaction()

2742 else:

2743 raise exc.InvalidRequestError("This transaction is inactive")

2744

2745 assert not self.is_active

2746 assert self.connection._transaction is not self

2747

2748

2749class NestedTransaction(Transaction):

2750 """Represent a 'nested', or SAVEPOINT transaction.

2751

2752 The :class:`.NestedTransaction` object is created by calling the

2753 :meth:`_engine.Connection.begin_nested` method of

2754 :class:`_engine.Connection`.

2755

2756 When using :class:`.NestedTransaction`, the semantics of "begin" /

2757 "commit" / "rollback" are as follows:

2758

2759 * the "begin" operation corresponds to the "BEGIN SAVEPOINT" command, where

2760 the savepoint is given an explicit name that is part of the state

2761 of this object.

2762

2763 * The :meth:`.NestedTransaction.commit` method corresponds to a

2764 "RELEASE SAVEPOINT" operation, using the savepoint identifier associated

2765 with this :class:`.NestedTransaction`.

2766

2767 * The :meth:`.NestedTransaction.rollback` method corresponds to a

2768 "ROLLBACK TO SAVEPOINT" operation, using the savepoint identifier

2769 associated with this :class:`.NestedTransaction`.

2770

2771 The rationale for mimicking the semantics of an outer transaction in

2772 terms of savepoints so that code may deal with a "savepoint" transaction

2773 and an "outer" transaction in an agnostic way.

2774

2775 .. seealso::

2776

2777 :ref:`session_begin_nested` - ORM version of the SAVEPOINT API.

2778

2779 """

2780

2781 __slots__ = ("connection", "is_active", "_savepoint", "_previous_nested")

2782

2783 _savepoint: str

2784

2785 def __init__(self, connection: Connection):

2786 assert connection._transaction is not None

2787 if connection._trans_context_manager:

2788 TransactionalContext._trans_ctx_check(connection)

2789 self.connection = connection

2790 self._savepoint = self.connection._savepoint_impl()

2791 self.is_active = True

2792 self._previous_nested = connection._nested_transaction

2793 connection._nested_transaction = self

2794

2795 def _deactivate_from_connection(self, warn: bool = True) -> None:

2796 if self.connection._nested_transaction is self:

2797 self.connection._nested_transaction = self._previous_nested

2798 elif warn:

2799 util.warn(

2800 "nested transaction already deassociated from connection"

2801 )

2802

2803 @property

2804 def _deactivated_from_connection(self) -> bool:

2805 return self.connection._nested_transaction is not self

2806

2807 def _cancel(self) -> None:

2808 # called by RootTransaction when the outer transaction is

2809 # committed, rolled back, or closed to cancel all savepoints

2810 # without any action being taken

2811 self.is_active = False

2812 self._deactivate_from_connection()

2813 if self._previous_nested:

2814 self._previous_nested._cancel()

2815

2816 def _close_impl(

2817 self, deactivate_from_connection: bool, warn_already_deactive: bool

2818 ) -> None:

2819 try:

2820 if (

2821 self.is_active

2822 and self.connection._transaction

2823 and self.connection._transaction.is_active

2824 ):

2825 self.connection._rollback_to_savepoint_impl(self._savepoint)

2826 finally:

2827 self.is_active = False

2828

2829 if deactivate_from_connection:

2830 self._deactivate_from_connection(warn=warn_already_deactive)

2831

2832 assert not self.is_active

2833 if deactivate_from_connection:

2834 assert self.connection._nested_transaction is not self

2835

2836 def _do_close(self) -> None:

2837 self._close_impl(True, False)

2838

2839 def _do_rollback(self) -> None:

2840 self._close_impl(True, True)

2841

2842 def _do_commit(self) -> None:

2843 if self.is_active:

2844 try:

2845 self.connection._release_savepoint_impl(self._savepoint)

2846 finally:

2847 # nested trans becomes inactive on failed release

2848 # unconditionally. this prevents it from trying to

2849 # emit SQL when it rolls back.

2850 self.is_active = False

2851

2852 # but only de-associate from connection if it succeeded

2853 self._deactivate_from_connection()

2854 else:

2855 if self.connection._nested_transaction is self:

2856 self.connection._invalid_transaction()

2857 else:

2858 raise exc.InvalidRequestError(

2859 "This nested transaction is inactive"

2860 )

2861

2862

2863class TwoPhaseTransaction(RootTransaction):

2864 """Represent a two-phase transaction.

2865

2866 A new :class:`.TwoPhaseTransaction` object may be procured

2867 using the :meth:`_engine.Connection.begin_twophase` method.

2868

2869 The interface is the same as that of :class:`.Transaction`

2870 with the addition of the :meth:`prepare` method.

2871

2872 """

2873

2874 __slots__ = ("xid", "_is_prepared")

2875

2876 xid: Any

2877

2878 def __init__(self, connection: Connection, xid: Any):

2879 self._is_prepared = False

2880 self.xid = xid

2881 super().__init__(connection)

2882

2883 def prepare(self) -> None:

2884 """Prepare this :class:`.TwoPhaseTransaction`.

2885

2886 After a PREPARE, the transaction can be committed.

2887

2888 """

2889 if not self.is_active:

2890 raise exc.InvalidRequestError("This transaction is inactive")

2891 self.connection._prepare_twophase_impl(self.xid)

2892 self._is_prepared = True

2893

2894 def _connection_begin_impl(self) -> None:

2895 self.connection._begin_twophase_impl(self)

2896

2897 def _connection_rollback_impl(self) -> None:

2898 self.connection._rollback_twophase_impl(self.xid, self._is_prepared)

2899

2900 def _connection_commit_impl(self) -> None:

2901 self.connection._commit_twophase_impl(self.xid, self._is_prepared)

2902

2903

2904class Engine(

2905 ConnectionEventsTarget, log.Identified, inspection.Inspectable["Inspector"]

2906):

2907 """

2908 Connects a :class:`~sqlalchemy.pool.Pool` and

2909 :class:`~sqlalchemy.engine.interfaces.Dialect` together to provide a

2910 source of database connectivity and behavior.

2911

2912 An :class:`_engine.Engine` object is instantiated publicly using the

2913 :func:`~sqlalchemy.create_engine` function.

2914

2915 .. seealso::

2916

2917 :doc:`/core/engines`

2918

2919 :ref:`connections_toplevel`

2920

2921 """

2922

2923 dispatch: dispatcher[ConnectionEventsTarget]

2924

2925 _compiled_cache: Optional[CompiledCacheType]

2926

2927 _execution_options: _ExecuteOptions = _EMPTY_EXECUTION_OPTS

2928 _has_events: bool = False

2929 _connection_cls: Type[Connection] = Connection

2930 _sqla_logger_namespace: str = "sqlalchemy.engine.Engine"

2931 _is_future: bool = False

2932

2933 _schema_translate_map: Optional[SchemaTranslateMapType] = None

2934 _option_cls: Type[OptionEngine]

2935

2936 dialect: Dialect

2937 pool: Pool

2938 url: URL

2939 hide_parameters: bool

2940

2941 def __init__(

2942 self,

2943 pool: Pool,

2944 dialect: Dialect,

2945 url: URL,

2946 logging_name: Optional[str] = None,

2947 echo: Optional[_EchoFlagType] = None,

2948 query_cache_size: int = 500,

2949 execution_options: Optional[Mapping[str, Any]] = None,

2950 hide_parameters: bool = False,

2951 ):

2952 self.pool = pool

2953 self.url = url

2954 self.dialect = dialect

2955 if logging_name:

2956 self.logging_name = logging_name

2957 self.echo = echo

2958 self.hide_parameters = hide_parameters

2959 if query_cache_size != 0:

2960 self._compiled_cache = util.LRUCache(

2961 query_cache_size, size_alert=self._lru_size_alert

2962 )

2963 else:

2964 self._compiled_cache = None

2965 log.instance_logger(self, echoflag=echo)

2966 if execution_options:

2967 self.update_execution_options(**execution_options)

2968

2969 def _lru_size_alert(self, cache: util.LRUCache[Any, Any]) -> None:

2970 if self._should_log_info():

2971 self.logger.info(

2972 "Compiled cache size pruning from %d items to %d. "

2973 "Increase cache size to reduce the frequency of pruning.",

2974 len(cache),

2975 cache.capacity,

2976 )

2977

2978 @property

2979 def engine(self) -> Engine:

2980 """Returns this :class:`.Engine`.

2981

2982 Used for legacy schemes that accept :class:`.Connection` /

2983 :class:`.Engine` objects within the same variable.

2984

2985 """

2986 return self

2987

2988 def clear_compiled_cache(self) -> None:

2989 """Clear the compiled cache associated with the dialect.

2990

2991 This applies **only** to the built-in cache that is established

2992 via the :paramref:`_engine.create_engine.query_cache_size` parameter.

2993 It will not impact any dictionary caches that were passed via the

2994 :paramref:`.Connection.execution_options.compiled_cache` parameter.

2995

2996 .. versionadded:: 1.4

2997

2998 """

2999 if self._compiled_cache:

3000 self._compiled_cache.clear()

3001

3002 def update_execution_options(self, **opt: Any) -> None:

3003 r"""Update the default execution_options dictionary

3004 of this :class:`_engine.Engine`.

3005

3006 The given keys/values in \**opt are added to the

3007 default execution options that will be used for

3008 all connections. The initial contents of this dictionary

3009 can be sent via the ``execution_options`` parameter

3010 to :func:`_sa.create_engine`.

3011

3012 .. seealso::

3013

3014 :meth:`_engine.Connection.execution_options`

3015

3016 :meth:`_engine.Engine.execution_options`

3017

3018 """

3019 self.dispatch.set_engine_execution_options(self, opt)

3020 self._execution_options = self._execution_options.union(opt)

3021 self.dialect.set_engine_execution_options(self, opt)

3022

3023 @overload

3024 def execution_options(

3025 self,

3026 *,

3027 compiled_cache: Optional[CompiledCacheType] = ...,

3028 logging_token: str = ...,

3029 isolation_level: IsolationLevel = ...,

3030 insertmanyvalues_page_size: int = ...,

3031 schema_translate_map: Optional[SchemaTranslateMapType] = ...,

3032 **opt: Any,

3033 ) -> OptionEngine: ...

3034

3035 @overload

3036 def execution_options(self, **opt: Any) -> OptionEngine: ...

3037

3038 def execution_options(self, **opt: Any) -> OptionEngine:

3039 """Return a new :class:`_engine.Engine` that will provide

3040 :class:`_engine.Connection` objects with the given execution options.

3041

3042 The returned :class:`_engine.Engine` remains related to the original

3043 :class:`_engine.Engine` in that it shares the same connection pool and

3044 other state:

3045

3046 * The :class:`_pool.Pool` used by the new :class:`_engine.Engine`

3047 is the

3048 same instance. The :meth:`_engine.Engine.dispose`

3049 method will replace

3050 the connection pool instance for the parent engine as well

3051 as this one.

3052 * Event listeners are "cascaded" - meaning, the new

3053 :class:`_engine.Engine`

3054 inherits the events of the parent, and new events can be associated

3055 with the new :class:`_engine.Engine` individually.

3056 * The logging configuration and logging_name is copied from the parent

3057 :class:`_engine.Engine`.

3058

3059 The intent of the :meth:`_engine.Engine.execution_options` method is

3060 to implement schemes where multiple :class:`_engine.Engine`

3061 objects refer to the same connection pool, but are differentiated

3062 by options that affect some execution-level behavior for each

3063 engine. One such example is breaking into separate "reader" and

3064 "writer" :class:`_engine.Engine` instances, where one

3065 :class:`_engine.Engine`

3066 has a lower :term:`isolation level` setting configured or is even

3067 transaction-disabled using "autocommit". An example of this

3068 configuration is at :ref:`dbapi_autocommit_multiple`.

3069

3070 Another example is one that

3071 uses a custom option ``shard_id`` which is consumed by an event

3072 to change the current schema on a database connection::

3073

3074 from sqlalchemy import event

3075 from sqlalchemy.engine import Engine

3076

3077 primary_engine = create_engine("mysql+mysqldb://")

3078 shard1 = primary_engine.execution_options(shard_id="shard1")

3079 shard2 = primary_engine.execution_options(shard_id="shard2")

3080

3081 shards = {"default": "base", "shard_1": "db1", "shard_2": "db2"}

3082

3083

3084 @event.listens_for(Engine, "before_cursor_execute")

3085 def _switch_shard(conn, cursor, stmt, params, context, executemany):

3086 shard_id = conn.get_execution_options().get("shard_id", "default")

3087 current_shard = conn.info.get("current_shard", None)

3088

3089 if current_shard != shard_id:

3090 cursor.execute("use %s" % shards[shard_id])

3091 conn.info["current_shard"] = shard_id

3092

3093 The above recipe illustrates two :class:`_engine.Engine` objects that

3094 will each serve as factories for :class:`_engine.Connection` objects

3095 that have pre-established "shard_id" execution options present. A

3096 :meth:`_events.ConnectionEvents.before_cursor_execute` event handler

3097 then interprets this execution option to emit a MySQL ``use`` statement

3098 to switch databases before a statement execution, while at the same

3099 time keeping track of which database we've established using the

3100 :attr:`_engine.Connection.info` dictionary.

3101

3102 .. seealso::

3103

3104 :meth:`_engine.Connection.execution_options`

3105 - update execution options

3106 on a :class:`_engine.Connection` object.

3107

3108 :meth:`_engine.Engine.update_execution_options`

3109 - update the execution

3110 options for a given :class:`_engine.Engine` in place.

3111

3112 :meth:`_engine.Engine.get_execution_options`

3113

3114

3115 """ # noqa: E501

3116 return self._option_cls(self, opt)

3117

3118 def get_execution_options(self) -> _ExecuteOptions:

3119 """Get the non-SQL options which will take effect during execution.

3120

3121 .. seealso::

3122

3123 :meth:`_engine.Engine.execution_options`

3124 """

3125 return self._execution_options

3126

3127 @property

3128 def name(self) -> str:

3129 """String name of the :class:`~sqlalchemy.engine.interfaces.Dialect`

3130 in use by this :class:`Engine`.

3131

3132 """

3133

3134 return self.dialect.name

3135

3136 @property

3137 def driver(self) -> str:

3138 """Driver name of the :class:`~sqlalchemy.engine.interfaces.Dialect`

3139 in use by this :class:`Engine`.

3140

3141 """

3142

3143 return self.dialect.driver

3144

3145 echo = log.echo_property()

3146

3147 def __repr__(self) -> str:

3148 return "Engine(%r)" % (self.url,)

3149

3150 def dispose(self, close: bool = True) -> None:

3151 """Dispose of the connection pool used by this

3152 :class:`_engine.Engine`.

3153

3154 A new connection pool is created immediately after the old one has been

3155 disposed. The previous connection pool is disposed either actively, by

3156 closing out all currently checked-in connections in that pool, or

3157 passively, by losing references to it but otherwise not closing any

3158 connections. The latter strategy is more appropriate for an initializer

3159 in a forked Python process.

3160

3161 Event listeners associated with the old pool via :class:`.PoolEvents`

3162 are **transferred to the new pool**; this is to support the pattern

3163 by which :class:`.PoolEvents` are set up in terms of the owning

3164 :class:`.Engine` without the need to refer to the :class:`.Pool`

3165 directly.

3166

3167 :param close: if left at its default of ``True``, has the

3168 effect of fully closing all **currently checked in**

3169 database connections. Connections that are still checked out

3170 will **not** be closed, however they will no longer be associated

3171 with this :class:`_engine.Engine`,

3172 so when they are closed individually, eventually the

3173 :class:`_pool.Pool` which they are associated with will

3174 be garbage collected and they will be closed out fully, if

3175 not already closed on checkin.

3176

3177 If set to ``False``, the previous connection pool is de-referenced,

3178 and otherwise not touched in any way.

3179

3180 .. versionadded:: 1.4.33 Added the :paramref:`.Engine.dispose.close`

3181 parameter to allow the replacement of a connection pool in a child

3182 process without interfering with the connections used by the parent

3183 process.

3184

3185

3186 .. seealso::

3187

3188 :ref:`engine_disposal`

3189

3190 :ref:`pooling_multiprocessing`

3191

3192 :meth:`.ConnectionEvents.engine_disposed`

3193

3194 """

3195 if close:

3196 self.pool.dispose()

3197 self.pool = self.pool.recreate()

3198 self.dispatch.engine_disposed(self)

3199

3200 @contextlib.contextmanager

3201 def _optional_conn_ctx_manager(

3202 self, connection: Optional[Connection] = None

3203 ) -> Iterator[Connection]:

3204 if connection is None:

3205 with self.connect() as conn:

3206 yield conn

3207 else:

3208 yield connection

3209

3210 @contextlib.contextmanager

3211 def begin(self) -> Iterator[Connection]:

3212 """Return a context manager delivering a :class:`_engine.Connection`

3213 with a :class:`.Transaction` established.

3214

3215 E.g.::

3216

3217 with engine.begin() as conn:

3218 conn.execute(text("insert into table (x, y, z) values (1, 2, 3)"))

3219 conn.execute(text("my_special_procedure(5)"))

3220

3221 Upon successful operation, the :class:`.Transaction`

3222 is committed. If an error is raised, the :class:`.Transaction`

3223 is rolled back.

3224

3225 .. seealso::

3226

3227 :meth:`_engine.Engine.connect` - procure a

3228 :class:`_engine.Connection` from

3229 an :class:`_engine.Engine`.

3230

3231 :meth:`_engine.Connection.begin` - start a :class:`.Transaction`

3232 for a particular :class:`_engine.Connection`.

3233

3234 """ # noqa: E501

3235 with self.connect() as conn:

3236 with conn.begin():

3237 yield conn

3238

3239 def _run_ddl_visitor(

3240 self,

3241 visitorcallable: Type[InvokeDDLBase],

3242 element: SchemaVisitable,

3243 **kwargs: Any,

3244 ) -> None:

3245 with self.begin() as conn:

3246 conn._run_ddl_visitor(visitorcallable, element, **kwargs)

3247

3248 def connect(self) -> Connection:

3249 """Return a new :class:`_engine.Connection` object.

3250

3251 The :class:`_engine.Connection` acts as a Python context manager, so

3252 the typical use of this method looks like::

3253

3254 with engine.connect() as connection:

3255 connection.execute(text("insert into table values ('foo')"))

3256 connection.commit()

3257

3258 Where above, after the block is completed, the connection is "closed"

3259 and its underlying DBAPI resources are returned to the connection pool.

3260 This also has the effect of rolling back any transaction that

3261 was explicitly begun or was begun via autobegin, and will

3262 emit the :meth:`_events.ConnectionEvents.rollback` event if one was

3263 started and is still in progress.

3264

3265 .. seealso::

3266

3267 :meth:`_engine.Engine.begin`

3268

3269 """

3270

3271 return self._connection_cls(self)

3272

3273 def raw_connection(self) -> PoolProxiedConnection:

3274 """Return a "raw" DBAPI connection from the connection pool.

3275

3276 The returned object is a proxied version of the DBAPI

3277 connection object used by the underlying driver in use.

3278 The object will have all the same behavior as the real DBAPI

3279 connection, except that its ``close()`` method will result in the

3280 connection being returned to the pool, rather than being closed

3281 for real.

3282

3283 This method provides direct DBAPI connection access for

3284 special situations when the API provided by

3285 :class:`_engine.Connection`

3286 is not needed. When a :class:`_engine.Connection` object is already

3287 present, the DBAPI connection is available using

3288 the :attr:`_engine.Connection.connection` accessor.

3289

3290 .. seealso::

3291

3292 :ref:`dbapi_connections`

3293

3294 """

3295 return self.pool.connect()

3296

3297

3298class OptionEngineMixin(log.Identified):

3299 _sa_propagate_class_events = False

3300

3301 dispatch: dispatcher[ConnectionEventsTarget]

3302 _compiled_cache: Optional[CompiledCacheType]

3303 dialect: Dialect

3304 pool: Pool

3305 url: URL

3306 hide_parameters: bool

3307 echo: log.echo_property

3308

3309 def __init__(

3310 self, proxied: Engine, execution_options: CoreExecuteOptionsParameter

3311 ):

3312 self._proxied = proxied

3313 self.url = proxied.url

3314 self.dialect = proxied.dialect

3315 self.logging_name = proxied.logging_name

3316 self.echo = proxied.echo

3317 self._compiled_cache = proxied._compiled_cache

3318 self.hide_parameters = proxied.hide_parameters

3319 log.instance_logger(self, echoflag=self.echo)

3320

3321 # note: this will propagate events that are assigned to the parent

3322 # engine after this OptionEngine is created. Since we share

3323 # the events of the parent we also disallow class-level events

3324 # to apply to the OptionEngine class directly.

3325 #

3326 # the other way this can work would be to transfer existing

3327 # events only, using:

3328 # self.dispatch._update(proxied.dispatch)

3329 #

3330 # that might be more appropriate however it would be a behavioral

3331 # change for logic that assigns events to the parent engine and

3332 # would like it to take effect for the already-created sub-engine.

3333 self.dispatch = self.dispatch._join(proxied.dispatch)

3334

3335 self._execution_options = proxied._execution_options

3336 self.update_execution_options(**execution_options)

3337

3338 def update_execution_options(self, **opt: Any) -> None:

3339 raise NotImplementedError()

3340

3341 if not typing.TYPE_CHECKING:

3342 # https://github.com/python/typing/discussions/1095

3343

3344 @property

3345 def pool(self) -> Pool:

3346 return self._proxied.pool

3347

3348 @pool.setter

3349 def pool(self, pool: Pool) -> None:

3350 self._proxied.pool = pool

3351

3352 @property

3353 def _has_events(self) -> bool:

3354 return self._proxied._has_events or self.__dict__.get(

3355 "_has_events", False

3356 )

3357

3358 @_has_events.setter

3359 def _has_events(self, value: bool) -> None:

3360 self.__dict__["_has_events"] = value

3361

3362

3363class OptionEngine(OptionEngineMixin, Engine):

3364 def update_execution_options(self, **opt: Any) -> None:

3365 Engine.update_execution_options(self, **opt)

3366

3367

3368Engine._option_cls = OptionEngine