callsite parameter fix while operating under asyncio (#565)

* Support for `asyncio` correct callsite params

* Support for `asyncio` correct callsite params

* Support for `asyncio` correct callsite params

* Added `versionchanged` tags

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Typing additions and corrections

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Spelling corrections

* Simplified logic and removed unused code

* Corrected typing

* Ensure context is cleared even if logging failure

* Renamed async contextvar to match project naming convention

* De-cleverfied logic for easier reading

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

CHANGELOG.md

@@ -17,6 +17,8 @@ You can find our backwards-compatibility policy [here](https://github.com/hynek/
 
 ### Fixed
 
+- `AsyncBoundLogger` now properly supports callsite parameters.
+
 - The return value from `get_logger()` (a `BoundLoggerLazyProxy`) now passes `isinstance`-checks against `structlog.typing.BindableLogger` on Python 3.12.
   [#561](https://github.com/hynek/structlog/issues/561)
 

src/structlog/_frames.py

@@ -11,6 +11,7 @@
 from io import StringIO
 from types import FrameType
 
+from .contextvars import _ASYNC_CALLING_STACK
 from .typing import ExcInfo
 
 
@@ -50,7 +51,7 @@ def _find_first_app_frame_and_name(
         tuple of (frame, name)
     """
     ignores = ["structlog"] + (additional_ignores or [])
-    f = sys._getframe()
+    f = _ASYNC_CALLING_STACK.get(sys._getframe())
     name = f.f_globals.get("__name__") or "?"
     while any(tuple(name.startswith(i) for i in ignores)):
         if f.f_back is None:

src/structlog/_log_levels.py

@@ -17,6 +17,7 @@
 from typing import Any, Callable
 
 from ._base import BoundLoggerBase
+from .contextvars import _ASYNC_CALLING_STACK
 from .typing import EventDict, FilteringBoundLogger
 
 
@@ -91,16 +92,26 @@ def exception(
 async def aexception(
     self: FilteringBoundLogger, event: str, *args: Any, **kw: Any
 ) -> Any:
+    """
+    .. versionchanged:: 23.3.0
+       Callsite parameters are now also collected under asyncio.
+    """
     # Exception info has to be extracted this early, because it is no longer
     # available once control is passed to the executor.
     if kw.get("exc_info", True) is True:
         kw["exc_info"] = sys.exc_info()
 
+    scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back)  # type: ignore[arg-type]
     ctx = contextvars.copy_context()
-    return await asyncio.get_running_loop().run_in_executor(
-        None,
-        lambda: ctx.run(lambda: self.error(event, *args, **kw)),
-    )
+    try:
+        runner = await asyncio.get_running_loop().run_in_executor(
+            None,
+            lambda: ctx.run(lambda: self.error(event, *args, **kw)),
+        )
+    finally:
+        _ASYNC_CALLING_STACK.reset(scs_token)
+
+    return runner
 
 
 def make_filtering_bound_logger(min_level: int) -> type[FilteringBoundLogger]:
@@ -170,16 +181,24 @@ def meth(self: Any, event: str, *args: Any, **kw: Any) -> Any:
             return self._proxy_to_logger(name, event % args, **kw)
 
         async def ameth(self: Any, event: str, *args: Any, **kw: Any) -> Any:
+            """
+            .. versionchanged:: 23.3.0
+               Callsite parameters are now also collected under asyncio.
+            """
             if args:
                 event = event % args
 
+            scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back)  # type: ignore[arg-type]
             ctx = contextvars.copy_context()
-            await asyncio.get_running_loop().run_in_executor(
-                None,
-                lambda: ctx.run(
-                    lambda: self._proxy_to_logger(name, event, **kw)
-                ),
-            )
+            try:
+                await asyncio.get_running_loop().run_in_executor(
+                    None,
+                    lambda: ctx.run(
+                        lambda: self._proxy_to_logger(name, event, **kw)
+                    ),
+                )
+            finally:
+                _ASYNC_CALLING_STACK.reset(scs_token)
 
         meth.__name__ = name
         ameth.__name__ = f"a{name}"
@@ -199,17 +218,28 @@ def log(self: Any, level: int, event: str, *args: Any, **kw: Any) -> Any:
     async def alog(
         self: Any, level: int, event: str, *args: Any, **kw: Any
     ) -> Any:
+        """
+        .. versionchanged:: 23.3.0
+           Callsite parameters are now also collected under asyncio.
+        """
         if level < min_level:
             return None
         name = _LEVEL_TO_NAME[level]
         if args:
             event = event % args
 
+        scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back)  # type: ignore[arg-type]
         ctx = contextvars.copy_context()
-        return await asyncio.get_running_loop().run_in_executor(
-            None,
-            lambda: ctx.run(lambda: self._proxy_to_logger(name, event, **kw)),
-        )
+        try:
+            runner = await asyncio.get_running_loop().run_in_executor(
+                None,
+                lambda: ctx.run(
+                    lambda: self._proxy_to_logger(name, event, **kw)
+                ),
+            )
+        finally:
+            _ASYNC_CALLING_STACK.reset(scs_token)
+        return runner
 
     meths: dict[str, Callable[..., Any]] = {"log": log, "alog": alog}
     for lvl, name in _LEVEL_TO_NAME.items():

src/structlog/contextvars.py

@@ -11,6 +11,8 @@
 .. versionchanged:: 21.1.0
    Reimplemented without using a single dict as context carrier for improved
    isolation. Every key-value pair is a separate `contextvars.ContextVar` now.
+.. versionchanged:: 23.3.0
+   Callsite parameters are now also collected under asyncio.
 
 See :doc:`contextvars`.
 """
@@ -20,6 +22,7 @@
 import contextlib
 import contextvars
 
+from types import FrameType
 from typing import Any, Generator, Mapping
 
 import structlog
@@ -30,6 +33,10 @@
 STRUCTLOG_KEY_PREFIX = "structlog_"
 STRUCTLOG_KEY_PREFIX_LEN = len(STRUCTLOG_KEY_PREFIX)
 
+_ASYNC_CALLING_STACK: contextvars.ContextVar[
+    FrameType
+] = contextvars.ContextVar("_ASYNC_CALLING_STACK")
+
 # For proper isolation, we have to use a dict of ContextVars instead of a
 # single ContextVar with a dict.
 # See https://github.com/hynek/structlog/pull/302 for details.

src/structlog/processors.py

@@ -734,10 +734,6 @@ class CallsiteParameterAdder:
     dictionaries with information such as the function name, line number and
     filename that an event dictionary originated from.
 
-    .. warning::
-        This processor cannot detect the correct callsite for invocation of
-        async functions.
-
     If the event dictionary has an embedded `logging.LogRecord` object and did
     not originate from *structlog* then the callsite information will be
     determined from the `logging.LogRecord` object. For event dictionaries

src/structlog/stdlib.py

@@ -24,7 +24,7 @@
 from ._base import BoundLoggerBase
 from ._frames import _find_first_app_frame_and_name, _format_stack
 from ._log_levels import _LEVEL_TO_NAME, _NAME_TO_LEVEL, add_log_level
-from .contextvars import merge_contextvars
+from .contextvars import _ASYNC_CALLING_STACK, merge_contextvars
 from .exceptions import DropEvent
 from .processors import StackInfoRenderer
 from .typing import Context, EventDict, ExcInfo, Processor, WrappedLogger
@@ -587,13 +587,19 @@ async def _dispatch_to_sync(
     ) -> None:
         """
         Merge contextvars and log using the sync logger in a thread pool.
+        .. versionchanged:: 23.3.0
+           Callsite parameters are now also collected under asyncio.
         """
+        scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back.f_back)  # type: ignore[union-attr, arg-type, unused-ignore]
         ctx = contextvars.copy_context()
 
-        await asyncio.get_running_loop().run_in_executor(
-            self._executor,
-            lambda: ctx.run(lambda: meth(event, *args, **kw)),
-        )
+        try:
+            await asyncio.get_running_loop().run_in_executor(
+                self._executor,
+                lambda: ctx.run(lambda: meth(event, *args, **kw)),
+            )
+        finally:
+            _ASYNC_CALLING_STACK.reset(scs_token)
 
     async def debug(self, event: str, *args: Any, **kw: Any) -> None:
         await self._dispatch_to_sync(self.sync_bl.debug, event, args, kw)

tests/test_processors.py

@@ -823,12 +823,6 @@ def test_all_parameters(self) -> None:
         }
         assert self.parameter_strings == self.get_callsite_parameters().keys()
 
-    @pytest.mark.xfail(
-        reason=(
-            "CallsiteParameterAdder cannot "
-            "determine the callsite for async calls."
-        )
-    )
     @pytest.mark.asyncio()
     async def test_async(self) -> None:
         """
@@ -852,10 +846,14 @@ def __init__(self):
 
         callsite_params = self.get_callsite_parameters()
         await logger.info("baz")
+        logger_params = json.loads(string_io.getvalue())
 
-        assert {"event": "baz", **callsite_params} == json.loads(
-            string_io.getvalue()
-        )
+        # These are different when running under async
+        for key in ["thread", "thread_name"]:
+            callsite_params.pop(key)
+            logger_params.pop(key)
+
+        assert {"event": "baz", **callsite_params} == logger_params
 
     def test_additional_ignores(self, monkeypatch: pytest.MonkeyPatch) -> None:
         """