(self)
| 425 | pass |
| 426 | |
| 427 | def _do_get(self) -> ConnectionPoolEntry: |
| 428 | try: |
| 429 | if TYPE_CHECKING: |
| 430 | c = cast(ConnectionPoolEntry, self._conn.current()) |
| 431 | else: |
| 432 | c = self._conn.current() |
| 433 | if c: |
| 434 | return c |
| 435 | except AttributeError: |
| 436 | pass |
| 437 | c = self._create_connection() |
| 438 | self._conn.current = weakref.ref(c) |
| 439 | if len(self._all_conns) >= self.size: |
| 440 | self._cleanup() |
| 441 | self._all_conns.add(c) |
| 442 | return c |
| 443 | |
| 444 | def connect(self) -> PoolProxiedConnection: |
| 445 | # vendored from Pool to include the now removed use_threadlocal |
nothing calls this directly
no test coverage detected