(self: Any, *args: Any, **kwargs: Any)
| 107 | |
| 108 | @functools.wraps(func) |
| 109 | async def csot_wrapper(self: Any, *args: Any, **kwargs: Any) -> Any: |
| 110 | if get_timeout() is None: |
| 111 | timeout = self._timeout |
| 112 | if timeout is not None: |
| 113 | with _TimeoutContext(timeout): |
| 114 | return await func(self, *args, **kwargs) |
| 115 | return await func(self, *args, **kwargs) |
| 116 | else: |
| 117 | |
| 118 | @functools.wraps(func) |
nothing calls this directly
no test coverage detected