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