| 76 | |
| 77 | |
| 78 | class PoolConnectionProxy(connection._ConnectionProxy, |
| 79 | metaclass=PoolConnectionProxyMeta, |
| 80 | wrap=True): |
| 81 | |
| 82 | __slots__ = ('_con', '_holder') |
| 83 | |
| 84 | def __init__( |
| 85 | self, holder: PoolConnectionHolder, con: connection.Connection |
| 86 | ) -> None: |
| 87 | self._con = con |
| 88 | self._holder = holder |
| 89 | con._set_proxy(self) |
| 90 | |
| 91 | def __getattr__(self, attr: str) -> Any: |
| 92 | # Proxy all unresolved attributes to the wrapped Connection object. |
| 93 | return getattr(self._con, attr) |
| 94 | |
| 95 | def _detach(self) -> Optional[connection.Connection]: |
| 96 | if self._con is None: |
| 97 | return |
| 98 | |
| 99 | con, self._con = self._con, None |
| 100 | con._set_proxy(None) |
| 101 | return con |
| 102 | |
| 103 | def __repr__(self) -> str: |
| 104 | if self._con is None: |
| 105 | return '<{classname} [released] {id:#x}>'.format( |
| 106 | classname=self.__class__.__name__, id=id(self)) |
| 107 | else: |
| 108 | return '<{classname} {con!r} {id:#x}>'.format( |
| 109 | classname=self.__class__.__name__, con=self._con, id=id(self)) |
| 110 | |
| 111 | |
| 112 | class PoolConnectionHolder: |
no outgoing calls
no test coverage detected
searching dependent graphs…