Merge contextvars and log using the sync logger in a thread pool.
(
self,
meth: Callable[..., Any],
event: str,
args: tuple[Any, ...],
kw: dict[str, Any],
)
| 653 | ) |
| 654 | |
| 655 | async def _dispatch_to_sync( |
| 656 | self, |
| 657 | meth: Callable[..., Any], |
| 658 | event: str, |
| 659 | args: tuple[Any, ...], |
| 660 | kw: dict[str, Any], |
| 661 | ) -> None: |
| 662 | """ |
| 663 | Merge contextvars and log using the sync logger in a thread pool. |
| 664 | """ |
| 665 | # Capture thread-specific info before handing off to the executor. |
| 666 | thread_token = _ASYNC_CALLING_THREAD.set( |
| 667 | (threading.get_ident(), threading.current_thread().name) |
| 668 | ) |
| 669 | scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back.f_back) # type: ignore[union-attr, arg-type, unused-ignore] |
| 670 | ctx = contextvars.copy_context() |
| 671 | |
| 672 | try: |
| 673 | await asyncio.get_running_loop().run_in_executor( |
| 674 | self._executor, |
| 675 | lambda: ctx.run(lambda: meth(event, *args, **kw)), |
| 676 | ) |
| 677 | finally: |
| 678 | _ASYNC_CALLING_STACK.reset(scs_token) |
| 679 | _ASYNC_CALLING_THREAD.reset(thread_token) |
| 680 | |
| 681 | async def debug(self, event: str, *args: Any, **kw: Any) -> None: |
| 682 | await self._dispatch_to_sync(self.sync_bl.debug, event, args, kw) |