(
mcls,
name: str,
bases: tuple[Type[Any], ...],
dct: dict[str, Any],
*,
wrap: bool = False,
)
| 28 | class PoolConnectionProxyMeta(type): |
| 29 | |
| 30 | def __new__( |
| 31 | mcls, |
| 32 | name: str, |
| 33 | bases: tuple[Type[Any], ...], |
| 34 | dct: dict[str, Any], |
| 35 | *, |
| 36 | wrap: bool = False, |
| 37 | ) -> PoolConnectionProxyMeta: |
| 38 | if wrap: |
| 39 | for attrname in dir(connection.Connection): |
| 40 | if attrname.startswith('_') or attrname in dct: |
| 41 | continue |
| 42 | |
| 43 | meth = getattr(connection.Connection, attrname) |
| 44 | if not inspect.isfunction(meth): |
| 45 | continue |
| 46 | |
| 47 | iscoroutine = inspect.iscoroutinefunction(meth) |
| 48 | wrapper = mcls._wrap_connection_method(attrname, iscoroutine) |
| 49 | wrapper = functools.update_wrapper(wrapper, meth) |
| 50 | dct[attrname] = wrapper |
| 51 | |
| 52 | if '__doc__' not in dct: |
| 53 | dct['__doc__'] = connection.Connection.__doc__ |
| 54 | |
| 55 | return super().__new__(mcls, name, bases, dct) |
| 56 | |
| 57 | @staticmethod |
| 58 | def _wrap_connection_method( |
nothing calls this directly
no test coverage detected