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

1# pool/base.py 

2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# 

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

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

7 

8 

9"""Base constructs for connection pools.""" 

10 

11from __future__ import annotations 

12 

13from collections import deque 

14import dataclasses 

15from enum import Enum 

16import threading 

17import time 

18import typing 

19from typing import Any 

20from typing import Callable 

21from typing import cast 

22from typing import Deque 

23from typing import Dict 

24from typing import List 

25from typing import Optional 

26from typing import Tuple 

27from typing import TYPE_CHECKING 

28from typing import Union 

29import weakref 

30 

31from .. import event 

32from .. import exc 

33from .. import log 

34from .. import util 

35from ..util.typing import Literal 

36from ..util.typing import Protocol 

37from ..util.typing import Self 

38 

39if TYPE_CHECKING: 

40 from ..engine.interfaces import DBAPIConnection 

41 from ..engine.interfaces import DBAPICursor 

42 from ..engine.interfaces import Dialect 

43 from ..event import _DispatchCommon 

44 from ..event import _ListenerFnType 

45 from ..event import dispatcher 

46 from ..sql._typing import _InfoType 

47 

48 

49@dataclasses.dataclass(frozen=True) 

50class PoolResetState: 

51 """describes the state of a DBAPI connection as it is being passed to 

52 the :meth:`.PoolEvents.reset` connection pool event. 

53 

54 .. versionadded:: 2.0.0b3 

55 

56 """ 

57 

58 __slots__ = ("transaction_was_reset", "terminate_only", "asyncio_safe") 

59 

60 transaction_was_reset: bool 

61 """Indicates if the transaction on the DBAPI connection was already 

62 essentially "reset" back by the :class:`.Connection` object. 

63 

64 This boolean is True if the :class:`.Connection` had transactional 

65 state present upon it, which was then not closed using the 

66 :meth:`.Connection.rollback` or :meth:`.Connection.commit` method; 

67 instead, the transaction was closed inline within the 

68 :meth:`.Connection.close` method so is guaranteed to remain non-present 

69 when this event is reached. 

70 

71 """ 

72 

73 terminate_only: bool 

74 """indicates if the connection is to be immediately terminated and 

75 not checked in to the pool. 

76 

77 This occurs for connections that were invalidated, as well as asyncio 

78 connections that were not cleanly handled by the calling code that 

79 are instead being garbage collected. In the latter case, 

80 operations can't be safely run on asyncio connections within garbage 

81 collection as there is not necessarily an event loop present. 

82 

83 """ 

84 

85 asyncio_safe: bool 

86 """Indicates if the reset operation is occurring within a scope where 

87 an enclosing event loop is expected to be present for asyncio applications. 

88 

89 Will be False in the case that the connection is being garbage collected. 

90 

91 """ 

92 

93 

94class ResetStyle(Enum): 

95 """Describe options for "reset on return" behaviors.""" 

96 

97 reset_rollback = 0 

98 reset_commit = 1 

99 reset_none = 2 

100 

101 

102_ResetStyleArgType = Union[ 

103 ResetStyle, 

104 Literal[True, None, False, "commit", "rollback"], 

105] 

106reset_rollback, reset_commit, reset_none = list(ResetStyle) 

107 

108 

109class _ConnDialect: 

110 """partial implementation of :class:`.Dialect` 

111 which provides DBAPI connection methods. 

112 

113 When a :class:`_pool.Pool` is combined with an :class:`_engine.Engine`, 

114 the :class:`_engine.Engine` replaces this with its own 

115 :class:`.Dialect`. 

116 

117 """ 

118 

119 is_async = False 

120 has_terminate = False 

121 

122 def do_rollback(self, dbapi_connection: PoolProxiedConnection) -> None: 

123 dbapi_connection.rollback() 

124 

125 def do_commit(self, dbapi_connection: PoolProxiedConnection) -> None: 

126 dbapi_connection.commit() 

127 

128 def do_terminate(self, dbapi_connection: DBAPIConnection) -> None: 

129 dbapi_connection.close() 

130 

131 def do_close(self, dbapi_connection: DBAPIConnection) -> None: 

132 dbapi_connection.close() 

133 

134 def _do_ping_w_event(self, dbapi_connection: DBAPIConnection) -> bool: 

135 raise NotImplementedError( 

136 "The ping feature requires that a dialect is " 

137 "passed to the connection pool." 

138 ) 

139 

140 def get_driver_connection(self, connection: DBAPIConnection) -> Any: 

141 return connection 

142 

143 

144class _AsyncConnDialect(_ConnDialect): 

145 is_async = True 

146 

147 

148class _CreatorFnType(Protocol): 

149 def __call__(self) -> DBAPIConnection: ... 

150 

151 

152class _CreatorWRecFnType(Protocol): 

153 def __call__(self, rec: ConnectionPoolEntry) -> DBAPIConnection: ... 

154 

155 

156class Pool(log.Identified, event.EventTarget): 

157 """Abstract base class for connection pools.""" 

158 

159 dispatch: dispatcher[Pool] 

160 echo: log._EchoFlagType 

161 

162 _orig_logging_name: Optional[str] 

163 _dialect: Union[_ConnDialect, Dialect] = _ConnDialect() 

164 _creator_arg: Union[_CreatorFnType, _CreatorWRecFnType] 

165 _invoke_creator: _CreatorWRecFnType 

166 _invalidate_time: float 

167 

168 def __init__( 

169 self, 

170 creator: Union[_CreatorFnType, _CreatorWRecFnType], 

171 recycle: int = -1, 

172 echo: log._EchoFlagType = None, 

173 logging_name: Optional[str] = None, 

174 reset_on_return: _ResetStyleArgType = True, 

175 events: Optional[List[Tuple[_ListenerFnType, str]]] = None, 

176 dialect: Optional[Union[_ConnDialect, Dialect]] = None, 

177 pre_ping: bool = False, 

178 _dispatch: Optional[_DispatchCommon[Pool]] = None, 

179 ): 

180 """ 

181 Construct a Pool. 

182 

183 :param creator: a callable function that returns a DB-API 

184 connection object. The function will be called with 

185 parameters. 

186 

187 :param recycle: If set to a value other than -1, number of 

188 seconds between connection recycling, which means upon 

189 checkout, if this timeout is surpassed the connection will be 

190 closed and replaced with a newly opened connection. Defaults to -1. 

191 

192 :param logging_name: String identifier which will be used within 

193 the "name" field of logging records generated within the 

194 "sqlalchemy.pool" logger. Defaults to a hexstring of the object's 

195 id. 

196 

197 :param echo: if True, the connection pool will log 

198 informational output such as when connections are invalidated 

199 as well as when connections are recycled to the default log handler, 

200 which defaults to ``sys.stdout`` for output.. If set to the string 

201 ``"debug"``, the logging will include pool checkouts and checkins. 

202 

203 The :paramref:`_pool.Pool.echo` parameter can also be set from the 

204 :func:`_sa.create_engine` call by using the 

205 :paramref:`_sa.create_engine.echo_pool` parameter. 

206 

207 .. seealso:: 

208 

209 :ref:`dbengine_logging` - further detail on how to configure 

210 logging. 

211 

212 :param reset_on_return: Determine steps to take on 

213 connections as they are returned to the pool, which were 

214 not otherwise handled by a :class:`_engine.Connection`. 

215 Available from :func:`_sa.create_engine` via the 

216 :paramref:`_sa.create_engine.pool_reset_on_return` parameter. 

217 

218 :paramref:`_pool.Pool.reset_on_return` can have any of these values: 

219 

220 * ``"rollback"`` - call rollback() on the connection, 

221 to release locks and transaction resources. 

222 This is the default value. The vast majority 

223 of use cases should leave this value set. 

224 * ``"commit"`` - call commit() on the connection, 

225 to release locks and transaction resources. 

226 A commit here may be desirable for databases that 

227 cache query plans if a commit is emitted, 

228 such as Microsoft SQL Server. However, this 

229 value is more dangerous than 'rollback' because 

230 any data changes present on the transaction 

231 are committed unconditionally. 

232 * ``None`` - don't do anything on the connection. 

233 This setting may be appropriate if the database / DBAPI 

234 works in pure "autocommit" mode at all times, or if 

235 a custom reset handler is established using the 

236 :meth:`.PoolEvents.reset` event handler. 

237 

238 * ``True`` - same as 'rollback', this is here for 

239 backwards compatibility. 

240 * ``False`` - same as None, this is here for 

241 backwards compatibility. 

242 

243 For further customization of reset on return, the 

244 :meth:`.PoolEvents.reset` event hook may be used which can perform 

245 any connection activity desired on reset. 

246 

247 .. seealso:: 

248 

249 :ref:`pool_reset_on_return` 

250 

251 :meth:`.PoolEvents.reset` 

252 

253 :param events: a list of 2-tuples, each of the form 

254 ``(callable, target)`` which will be passed to :func:`.event.listen` 

255 upon construction. Provided here so that event listeners 

256 can be assigned via :func:`_sa.create_engine` before dialect-level 

257 listeners are applied. 

258 

259 :param dialect: a :class:`.Dialect` that will handle the job 

260 of calling rollback(), close(), or commit() on DBAPI connections. 

261 If omitted, a built-in "stub" dialect is used. Applications that 

262 make use of :func:`_sa.create_engine` should not use this parameter 

263 as it is handled by the engine creation strategy. 

264 

265 :param pre_ping: if True, the pool will emit a "ping" (typically 

266 "SELECT 1", but is dialect-specific) on the connection 

267 upon checkout, to test if the connection is alive or not. If not, 

268 the connection is transparently re-connected and upon success, all 

269 other pooled connections established prior to that timestamp are 

270 invalidated. Requires that a dialect is passed as well to 

271 interpret the disconnection error. 

272 

273 .. versionadded:: 1.2 

274 

275 """ 

276 if logging_name: 

277 self.logging_name = self._orig_logging_name = logging_name 

278 else: 

279 self._orig_logging_name = None 

280 

281 log.instance_logger(self, echoflag=echo) 

282 self._creator = creator 

283 self._recycle = recycle 

284 self._invalidate_time = 0 

285 self._pre_ping = pre_ping 

286 self._reset_on_return = util.parse_user_argument_for_enum( 

287 reset_on_return, 

288 { 

289 ResetStyle.reset_rollback: ["rollback", True], 

290 ResetStyle.reset_none: ["none", None, False], 

291 ResetStyle.reset_commit: ["commit"], 

292 }, 

293 "reset_on_return", 

294 ) 

295 

296 self.echo = echo 

297 

298 if _dispatch: 

299 self.dispatch._update(_dispatch, only_propagate=False) 

300 if dialect: 

301 self._dialect = dialect 

302 if events: 

303 for fn, target in events: 

304 event.listen(self, target, fn) 

305 

306 @util.hybridproperty 

307 def _is_asyncio(self) -> bool: 

308 return self._dialect.is_async 

309 

310 @property 

311 def _creator(self) -> Union[_CreatorFnType, _CreatorWRecFnType]: 

312 return self._creator_arg 

313 

314 @_creator.setter 

315 def _creator( 

316 self, creator: Union[_CreatorFnType, _CreatorWRecFnType] 

317 ) -> None: 

318 self._creator_arg = creator 

319 

320 # mypy seems to get super confused assigning functions to 

321 # attributes 

322 self._invoke_creator = self._should_wrap_creator(creator) 

323 

324 @_creator.deleter 

325 def _creator(self) -> None: 

326 # needed for mock testing 

327 del self._creator_arg 

328 del self._invoke_creator 

329 

330 def _should_wrap_creator( 

331 self, creator: Union[_CreatorFnType, _CreatorWRecFnType] 

332 ) -> _CreatorWRecFnType: 

333 """Detect if creator accepts a single argument, or is sent 

334 as a legacy style no-arg function. 

335 

336 """ 

337 

338 try: 

339 argspec = util.get_callable_argspec(self._creator, no_self=True) 

340 except TypeError: 

341 creator_fn = cast(_CreatorFnType, creator) 

342 return lambda rec: creator_fn() 

343 

344 if argspec.defaults is not None: 

345 defaulted = len(argspec.defaults) 

346 else: 

347 defaulted = 0 

348 positionals = len(argspec[0]) - defaulted 

349 

350 # look for the exact arg signature that DefaultStrategy 

351 # sends us 

352 if (argspec[0], argspec[3]) == (["connection_record"], (None,)): 

353 return cast(_CreatorWRecFnType, creator) 

354 # or just a single positional 

355 elif positionals == 1: 

356 return cast(_CreatorWRecFnType, creator) 

357 # all other cases, just wrap and assume legacy "creator" callable 

358 # thing 

359 else: 

360 creator_fn = cast(_CreatorFnType, creator) 

361 return lambda rec: creator_fn() 

362 

363 def _close_connection( 

364 self, connection: DBAPIConnection, *, terminate: bool = False 

365 ) -> None: 

366 self.logger.debug( 

367 "%s connection %r", 

368 "Hard-closing" if terminate else "Closing", 

369 connection, 

370 ) 

371 try: 

372 if terminate: 

373 self._dialect.do_terminate(connection) 

374 else: 

375 self._dialect.do_close(connection) 

376 except BaseException as e: 

377 self.logger.error( 

378 f"Exception {'terminating' if terminate else 'closing'} " 

379 f"connection %r", 

380 connection, 

381 exc_info=True, 

382 ) 

383 if not isinstance(e, Exception): 

384 raise 

385 

386 def _create_connection(self) -> ConnectionPoolEntry: 

387 """Called by subclasses to create a new ConnectionRecord.""" 

388 

389 return _ConnectionRecord(self) 

390 

391 def _invalidate( 

392 self, 

393 connection: PoolProxiedConnection, 

394 exception: Optional[BaseException] = None, 

395 _checkin: bool = True, 

396 ) -> None: 

397 """Mark all connections established within the generation 

398 of the given connection as invalidated. 

399 

400 If this pool's last invalidate time is before when the given 

401 connection was created, update the timestamp til now. Otherwise, 

402 no action is performed. 

403 

404 Connections with a start time prior to this pool's invalidation 

405 time will be recycled upon next checkout. 

406 """ 

407 rec = getattr(connection, "_connection_record", None) 

408 if not rec or self._invalidate_time < rec.starttime: 

409 self._invalidate_time = time.time() 

410 if _checkin and getattr(connection, "is_valid", False): 

411 connection.invalidate(exception) 

412 

413 def recreate(self) -> Pool: 

414 """Return a new :class:`_pool.Pool`, of the same class as this one 

415 and configured with identical creation arguments. 

416 

417 This method is used in conjunction with :meth:`dispose` 

418 to close out an entire :class:`_pool.Pool` and create a new one in 

419 its place. 

420 

421 """ 

422 

423 raise NotImplementedError() 

424 

425 def dispose(self) -> None: 

426 """Dispose of this pool. 

427 

428 This method leaves the possibility of checked-out connections 

429 remaining open, as it only affects connections that are 

430 idle in the pool. 

431 

432 .. seealso:: 

433 

434 :meth:`Pool.recreate` 

435 

436 """ 

437 

438 raise NotImplementedError() 

439 

440 def connect(self) -> PoolProxiedConnection: 

441 """Return a DBAPI connection from the pool. 

442 

443 The connection is instrumented such that when its 

444 ``close()`` method is called, the connection will be returned to 

445 the pool. 

446 

447 """ 

448 return _ConnectionFairy._checkout(self) 

449 

450 def _return_conn(self, record: ConnectionPoolEntry) -> None: 

451 """Given a _ConnectionRecord, return it to the :class:`_pool.Pool`. 

452 

453 This method is called when an instrumented DBAPI connection 

454 has its ``close()`` method called. 

455 

456 """ 

457 self._do_return_conn(record) 

458 

459 def _do_get(self) -> ConnectionPoolEntry: 

460 """Implementation for :meth:`get`, supplied by subclasses.""" 

461 

462 raise NotImplementedError() 

463 

464 def _do_return_conn(self, record: ConnectionPoolEntry) -> None: 

465 """Implementation for :meth:`return_conn`, supplied by subclasses.""" 

466 

467 raise NotImplementedError() 

468 

469 def status(self) -> str: 

470 """Returns a brief description of the state of this pool.""" 

471 raise NotImplementedError() 

472 

473 

474class ManagesConnection: 

475 """Common base for the two connection-management interfaces 

476 :class:`.PoolProxiedConnection` and :class:`.ConnectionPoolEntry`. 

477 

478 These two objects are typically exposed in the public facing API 

479 via the connection pool event hooks, documented at :class:`.PoolEvents`. 

480 

481 .. versionadded:: 2.0 

482 

483 """ 

484 

485 __slots__ = () 

486 

487 dbapi_connection: Optional[DBAPIConnection] 

488 """A reference to the actual DBAPI connection being tracked. 

489 

490 This is a :pep:`249`-compliant object that for traditional sync-style 

491 dialects is provided by the third-party 

492 DBAPI implementation in use. For asyncio dialects, the implementation 

493 is typically an adapter object provided by the SQLAlchemy dialect 

494 itself; the underlying asyncio object is available via the 

495 :attr:`.ManagesConnection.driver_connection` attribute. 

496 

497 SQLAlchemy's interface for the DBAPI connection is based on the 

498 :class:`.DBAPIConnection` protocol object 

499 

500 .. seealso:: 

501 

502 :attr:`.ManagesConnection.driver_connection` 

503 

504 :ref:`faq_dbapi_connection` 

505 

506 """ 

507 

508 driver_connection: Optional[Any] 

509 """The "driver level" connection object as used by the Python 

510 DBAPI or database driver. 

511 

512 For traditional :pep:`249` DBAPI implementations, this object will 

513 be the same object as that of 

514 :attr:`.ManagesConnection.dbapi_connection`. For an asyncio database 

515 driver, this will be the ultimate "connection" object used by that 

516 driver, such as the ``asyncpg.Connection`` object which will not have 

517 standard pep-249 methods. 

518 

519 .. versionadded:: 1.4.24 

520 

521 .. seealso:: 

522 

523 :attr:`.ManagesConnection.dbapi_connection` 

524 

525 :ref:`faq_dbapi_connection` 

526 

527 """ 

528 

529 @util.ro_memoized_property 

530 def info(self) -> _InfoType: 

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

532 referred to by this :class:`.ManagesConnection` instance, allowing 

533 user-defined data to be associated with the connection. 

534 

535 The data in this dictionary is persistent for the lifespan 

536 of the DBAPI connection itself, including across pool checkins 

537 and checkouts. When the connection is invalidated 

538 and replaced with a new one, this dictionary is cleared. 

539 

540 For a :class:`.PoolProxiedConnection` instance that's not associated 

541 with a :class:`.ConnectionPoolEntry`, such as if it were detached, the 

542 attribute returns a dictionary that is local to that 

543 :class:`.ConnectionPoolEntry`. Therefore the 

544 :attr:`.ManagesConnection.info` attribute will always provide a Python 

545 dictionary. 

546 

547 .. seealso:: 

548 

549 :attr:`.ManagesConnection.record_info` 

550 

551 

552 """ 

553 raise NotImplementedError() 

554 

555 @util.ro_memoized_property 

556 def record_info(self) -> Optional[_InfoType]: 

557 """Persistent info dictionary associated with this 

558 :class:`.ManagesConnection`. 

559 

560 Unlike the :attr:`.ManagesConnection.info` dictionary, the lifespan 

561 of this dictionary is that of the :class:`.ConnectionPoolEntry` 

562 which owns it; therefore this dictionary will persist across 

563 reconnects and connection invalidation for a particular entry 

564 in the connection pool. 

565 

566 For a :class:`.PoolProxiedConnection` instance that's not associated 

567 with a :class:`.ConnectionPoolEntry`, such as if it were detached, the 

568 attribute returns None. Contrast to the :attr:`.ManagesConnection.info` 

569 dictionary which is never None. 

570 

571 

572 .. seealso:: 

573 

574 :attr:`.ManagesConnection.info` 

575 

576 """ 

577 raise NotImplementedError() 

578 

579 def invalidate( 

580 self, e: Optional[BaseException] = None, soft: bool = False 

581 ) -> None: 

582 """Mark the managed connection as invalidated. 

583 

584 :param e: an exception object indicating a reason for the invalidation. 

585 

586 :param soft: if True, the connection isn't closed; instead, this 

587 connection will be recycled on next checkout. 

588 

589 .. seealso:: 

590 

591 :ref:`pool_connection_invalidation` 

592 

593 

594 """ 

595 raise NotImplementedError() 

596 

597 

598class ConnectionPoolEntry(ManagesConnection): 

599 """Interface for the object that maintains an individual database 

600 connection on behalf of a :class:`_pool.Pool` instance. 

601 

602 The :class:`.ConnectionPoolEntry` object represents the long term 

603 maintenance of a particular connection for a pool, including expiring or 

604 invalidating that connection to have it replaced with a new one, which will 

605 continue to be maintained by that same :class:`.ConnectionPoolEntry` 

606 instance. Compared to :class:`.PoolProxiedConnection`, which is the 

607 short-term, per-checkout connection manager, this object lasts for the 

608 lifespan of a particular "slot" within a connection pool. 

609 

610 The :class:`.ConnectionPoolEntry` object is mostly visible to public-facing 

611 API code when it is delivered to connection pool event hooks, such as 

612 :meth:`_events.PoolEvents.connect` and :meth:`_events.PoolEvents.checkout`. 

613 

614 .. versionadded:: 2.0 :class:`.ConnectionPoolEntry` provides the public 

615 facing interface for the :class:`._ConnectionRecord` internal class. 

616 

617 """ 

618 

619 __slots__ = () 

620 

621 @property 

622 def in_use(self) -> bool: 

623 """Return True the connection is currently checked out""" 

624 

625 raise NotImplementedError() 

626 

627 def close(self) -> None: 

628 """Close the DBAPI connection managed by this connection pool entry.""" 

629 raise NotImplementedError() 

630 

631 

632class _ConnectionRecord(ConnectionPoolEntry): 

633 """Maintains a position in a connection pool which references a pooled 

634 connection. 

635 

636 This is an internal object used by the :class:`_pool.Pool` implementation 

637 to provide context management to a DBAPI connection maintained by 

638 that :class:`_pool.Pool`. The public facing interface for this class 

639 is described by the :class:`.ConnectionPoolEntry` class. See that 

640 class for public API details. 

641 

642 .. seealso:: 

643 

644 :class:`.ConnectionPoolEntry` 

645 

646 :class:`.PoolProxiedConnection` 

647 

648 """ 

649 

650 __slots__ = ( 

651 "__pool", 

652 "fairy_ref", 

653 "finalize_callback", 

654 "fresh", 

655 "starttime", 

656 "dbapi_connection", 

657 "__weakref__", 

658 "__dict__", 

659 ) 

660 

661 finalize_callback: Deque[Callable[[DBAPIConnection], None]] 

662 fresh: bool 

663 fairy_ref: Optional[weakref.ref[_ConnectionFairy]] 

664 starttime: float 

665 

666 def __init__(self, pool: Pool, connect: bool = True): 

667 self.fresh = False 

668 self.fairy_ref = None 

669 self.starttime = 0 

670 self.dbapi_connection = None 

671 

672 self.__pool = pool 

673 if connect: 

674 self.__connect() 

675 self.finalize_callback = deque() 

676 

677 dbapi_connection: Optional[DBAPIConnection] 

678 

679 @property 

680 def driver_connection(self) -> Optional[Any]: # type: ignore[override] # mypy#4125 # noqa: E501 

681 if self.dbapi_connection is None: 

682 return None 

683 else: 

684 return self.__pool._dialect.get_driver_connection( 

685 self.dbapi_connection 

686 ) 

687 

688 @property 

689 @util.deprecated( 

690 "2.0", 

691 "The _ConnectionRecord.connection attribute is deprecated; " 

692 "please use 'driver_connection'", 

693 ) 

694 def connection(self) -> Optional[DBAPIConnection]: 

695 return self.dbapi_connection 

696 

697 _soft_invalidate_time: float = 0 

698 

699 @util.ro_memoized_property 

700 def info(self) -> _InfoType: 

701 return {} 

702 

703 @util.ro_memoized_property 

704 def record_info(self) -> Optional[_InfoType]: 

705 return {} 

706 

707 @classmethod 

708 def checkout(cls, pool: Pool) -> _ConnectionFairy: 

709 if TYPE_CHECKING: 

710 rec = cast(_ConnectionRecord, pool._do_get()) 

711 else: 

712 rec = pool._do_get() 

713 

714 try: 

715 dbapi_connection = rec.get_connection() 

716 except BaseException as err: 

717 with util.safe_reraise(): 

718 rec._checkin_failed(err, _fairy_was_created=False) 

719 

720 # not reached, for code linters only 

721 raise 

722 

723 echo = pool._should_log_debug() 

724 fairy = _ConnectionFairy(pool, dbapi_connection, rec, echo) 

725 

726 rec.fairy_ref = ref = weakref.ref( 

727 fairy, 

728 lambda ref: ( 

729 _finalize_fairy( 

730 None, rec, pool, ref, echo, transaction_was_reset=False 

731 ) 

732 if _finalize_fairy is not None 

733 else None 

734 ), 

735 ) 

736 _strong_ref_connection_records[ref] = rec 

737 if echo: 

738 pool.logger.debug( 

739 "Connection %r checked out from pool", dbapi_connection 

740 ) 

741 return fairy 

742 

743 def _checkin_failed( 

744 self, err: BaseException, _fairy_was_created: bool = True 

745 ) -> None: 

746 self.invalidate(e=err) 

747 self.checkin( 

748 _fairy_was_created=_fairy_was_created, 

749 ) 

750 

751 def checkin(self, _fairy_was_created: bool = True) -> None: 

752 if self.fairy_ref is None and _fairy_was_created: 

753 # _fairy_was_created is False for the initial get connection phase; 

754 # meaning there was no _ConnectionFairy and we must unconditionally 

755 # do a checkin. 

756 # 

757 # otherwise, if fairy_was_created==True, if fairy_ref is None here 

758 # that means we were checked in already, so this looks like 

759 # a double checkin. 

760 util.warn("Double checkin attempted on %s" % self) 

761 return 

762 self.fairy_ref = None 

763 connection = self.dbapi_connection 

764 pool = self.__pool 

765 while self.finalize_callback: 

766 finalizer = self.finalize_callback.pop() 

767 if connection is not None: 

768 finalizer(connection) 

769 if pool.dispatch.checkin: 

770 pool.dispatch.checkin(connection, self) 

771 

772 pool._return_conn(self) 

773 

774 @property 

775 def in_use(self) -> bool: 

776 return self.fairy_ref is not None 

777 

778 @property 

779 def last_connect_time(self) -> float: 

780 return self.starttime 

781 

782 def close(self) -> None: 

783 if self.dbapi_connection is not None: 

784 self.__close() 

785 

786 def invalidate( 

787 self, e: Optional[BaseException] = None, soft: bool = False 

788 ) -> None: 

789 # already invalidated 

790 if self.dbapi_connection is None: 

791 return 

792 if soft: 

793 self.__pool.dispatch.soft_invalidate( 

794 self.dbapi_connection, self, e 

795 ) 

796 else: 

797 self.__pool.dispatch.invalidate(self.dbapi_connection, self, e) 

798 if e is not None: 

799 self.__pool.logger.info( 

800 "%sInvalidate connection %r (reason: %s:%s)", 

801 "Soft " if soft else "", 

802 self.dbapi_connection, 

803 e.__class__.__name__, 

804 e, 

805 ) 

806 else: 

807 self.__pool.logger.info( 

808 "%sInvalidate connection %r", 

809 "Soft " if soft else "", 

810 self.dbapi_connection, 

811 ) 

812 

813 if soft: 

814 self._soft_invalidate_time = time.time() 

815 else: 

816 self.__close(terminate=True) 

817 self.dbapi_connection = None 

818 

819 def get_connection(self) -> DBAPIConnection: 

820 recycle = False 

821 

822 # NOTE: the various comparisons here are assuming that measurable time 

823 # passes between these state changes. however, time.time() is not 

824 # guaranteed to have sub-second precision. comparisons of 

825 # "invalidation time" to "starttime" should perhaps use >= so that the 

826 # state change can take place assuming no measurable time has passed, 

827 # however this does not guarantee correct behavior here as if time 

828 # continues to not pass, it will try to reconnect repeatedly until 

829 # these timestamps diverge, so in that sense using > is safer. Per 

830 # https://stackoverflow.com/a/1938096/34549, Windows time.time() may be 

831 # within 16 milliseconds accuracy, so unit tests for connection 

832 # invalidation need a sleep of at least this long between initial start 

833 # time and invalidation for the logic below to work reliably. 

834 

835 if self.dbapi_connection is None: 

836 self.info.clear() 

837 self.__connect() 

838 elif ( 

839 self.__pool._recycle > -1 

840 and time.time() - self.starttime > self.__pool._recycle 

841 ): 

842 self.__pool.logger.info( 

843 "Connection %r exceeded timeout; recycling", 

844 self.dbapi_connection, 

845 ) 

846 recycle = True 

847 elif self.__pool._invalidate_time > self.starttime: 

848 self.__pool.logger.info( 

849 "Connection %r invalidated due to pool invalidation; " 

850 + "recycling", 

851 self.dbapi_connection, 

852 ) 

853 recycle = True 

854 elif self._soft_invalidate_time > self.starttime: 

855 self.__pool.logger.info( 

856 "Connection %r invalidated due to local soft invalidation; " 

857 + "recycling", 

858 self.dbapi_connection, 

859 ) 

860 recycle = True 

861 

862 if recycle: 

863 self.__close(terminate=True) 

864 self.info.clear() 

865 

866 self.__connect() 

867 

868 assert self.dbapi_connection is not None 

869 return self.dbapi_connection 

870 

871 def _is_hard_or_soft_invalidated(self) -> bool: 

872 return ( 

873 self.dbapi_connection is None 

874 or self.__pool._invalidate_time > self.starttime 

875 or (self._soft_invalidate_time > self.starttime) 

876 ) 

877 

878 def __close(self, *, terminate: bool = False) -> None: 

879 self.finalize_callback.clear() 

880 if self.__pool.dispatch.close: 

881 self.__pool.dispatch.close(self.dbapi_connection, self) 

882 assert self.dbapi_connection is not None 

883 self.__pool._close_connection( 

884 self.dbapi_connection, terminate=terminate 

885 ) 

886 self.dbapi_connection = None 

887 

888 def __connect(self) -> None: 

889 pool = self.__pool 

890 

891 # ensure any existing connection is removed, so that if 

892 # creator fails, this attribute stays None 

893 self.dbapi_connection = None 

894 try: 

895 self.starttime = time.time() 

896 self.dbapi_connection = connection = pool._invoke_creator(self) 

897 pool.logger.debug("Created new connection %r", connection) 

898 self.fresh = True 

899 except BaseException as e: 

900 with util.safe_reraise(): 

901 pool.logger.debug("Error on connect(): %s", e) 

902 else: 

903 # in SQLAlchemy 1.4 the first_connect event is not used by 

904 # the engine, so this will usually not be set 

905 if pool.dispatch.first_connect: 

906 pool.dispatch.first_connect.for_modify( 

907 pool.dispatch 

908 ).exec_once_unless_exception(self.dbapi_connection, self) 

909 

910 # init of the dialect now takes place within the connect 

911 # event, so ensure a mutex is used on the first run 

912 pool.dispatch.connect.for_modify( 

913 pool.dispatch 

914 )._exec_w_sync_on_first_run(self.dbapi_connection, self) 

915 

916 

917def _finalize_fairy( 

918 dbapi_connection: Optional[DBAPIConnection], 

919 connection_record: Optional[_ConnectionRecord], 

920 pool: Pool, 

921 ref: Optional[ 

922 weakref.ref[_ConnectionFairy] 

923 ], # this is None when called directly, not by the gc 

924 echo: Optional[log._EchoFlagType], 

925 transaction_was_reset: bool = False, 

926 fairy: Optional[_ConnectionFairy] = None, 

927) -> None: 

928 """Cleanup for a :class:`._ConnectionFairy` whether or not it's already 

929 been garbage collected. 

930 

931 When using an async dialect no IO can happen here (without using 

932 a dedicated thread), since this is called outside the greenlet 

933 context and with an already running loop. In this case function 

934 will only log a message and raise a warning. 

935 """ 

936 

937 is_gc_cleanup = ref is not None 

938 

939 if is_gc_cleanup: 

940 assert ref is not None 

941 _strong_ref_connection_records.pop(ref, None) 

942 assert connection_record is not None 

943 if connection_record.fairy_ref is not ref: 

944 return 

945 assert dbapi_connection is None 

946 dbapi_connection = connection_record.dbapi_connection 

947 

948 elif fairy: 

949 _strong_ref_connection_records.pop(weakref.ref(fairy), None) 

950 

951 # null pool is not _is_asyncio but can be used also with async dialects 

952 dont_restore_gced = pool._dialect.is_async 

953 

954 if dont_restore_gced: 

955 detach = connection_record is None or is_gc_cleanup 

956 can_manipulate_connection = not is_gc_cleanup 

957 can_close_or_terminate_connection = ( 

958 not pool._dialect.is_async or pool._dialect.has_terminate 

959 ) 

960 requires_terminate_for_close = ( 

961 pool._dialect.is_async and pool._dialect.has_terminate 

962 ) 

963 

964 else: 

965 detach = connection_record is None 

966 can_manipulate_connection = can_close_or_terminate_connection = True 

967 requires_terminate_for_close = False 

968 

969 if dbapi_connection is not None: 

970 if connection_record and echo: 

971 pool.logger.debug( 

972 "Connection %r being returned to pool", dbapi_connection 

973 ) 

974 

975 try: 

976 if not fairy: 

977 assert connection_record is not None 

978 fairy = _ConnectionFairy( 

979 pool, 

980 dbapi_connection, 

981 connection_record, 

982 echo, 

983 ) 

984 assert fairy.dbapi_connection is dbapi_connection 

985 

986 fairy._reset( 

987 pool, 

988 transaction_was_reset=transaction_was_reset, 

989 terminate_only=detach, 

990 asyncio_safe=can_manipulate_connection, 

991 ) 

992 

993 if detach: 

994 if connection_record: 

995 fairy._pool = pool 

996 fairy.detach() 

997 

998 if can_close_or_terminate_connection: 

999 if pool.dispatch.close_detached: