diff --git a/src/structlog/_native.py b/src/structlog/_native.py index 5775b8de..e02a8e47 100644 --- a/src/structlog/_native.py +++ b/src/structlog/_native.py @@ -29,7 +29,12 @@ NOTSET, WARNING, ) -from .contextvars import _ASYNC_CALLING_STACK, _ASYNC_CALLING_THREAD +from .contextvars import ( + _ASYNC_CALLING_STACK, + _ASYNC_CALLING_TASK_NAME, + _ASYNC_CALLING_THREAD, + _get_current_task_name, +) from .typing import FilteringBoundLogger @@ -65,6 +70,7 @@ async def aexception( thread_token = _ASYNC_CALLING_THREAD.set( (threading.get_ident(), threading.current_thread().name) ) + task_name_token = _ASYNC_CALLING_TASK_NAME.set(_get_current_task_name()) scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back) # type: ignore[arg-type] ctx = contextvars.copy_context() @@ -76,6 +82,7 @@ async def aexception( finally: _ASYNC_CALLING_STACK.reset(scs_token) _ASYNC_CALLING_THREAD.reset(thread_token) + _ASYNC_CALLING_TASK_NAME.reset(task_name_token) return runner @@ -184,6 +191,9 @@ async def ameth(self: Any, event: str, *args: Any, **kw: Any) -> Any: thread_token = _ASYNC_CALLING_THREAD.set( (threading.get_ident(), threading.current_thread().name) ) + task_name_token = _ASYNC_CALLING_TASK_NAME.set( + _get_current_task_name() + ) scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back) # type: ignore[arg-type] ctx = contextvars.copy_context() @@ -197,6 +207,7 @@ async def ameth(self: Any, event: str, *args: Any, **kw: Any) -> Any: finally: _ASYNC_CALLING_STACK.reset(scs_token) _ASYNC_CALLING_THREAD.reset(thread_token) + _ASYNC_CALLING_TASK_NAME.reset(task_name_token) meth.__name__ = name ameth.__name__ = f"a{name}" @@ -228,6 +239,9 @@ async def alog( thread_token = _ASYNC_CALLING_THREAD.set( (threading.get_ident(), threading.current_thread().name) ) + task_name_token = _ASYNC_CALLING_TASK_NAME.set( + _get_current_task_name() + ) scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back) # type: ignore[arg-type] ctx = contextvars.copy_context() @@ -241,21 +255,28 @@ async def alog( finally: _ASYNC_CALLING_STACK.reset(scs_token) _ASYNC_CALLING_THREAD.reset(thread_token) + _ASYNC_CALLING_TASK_NAME.reset(task_name_token) return runner - meths: dict[str, Callable[..., Any]] = {"log": log, "alog": alog} + meths: dict[str, Callable[..., Any]] = { + "log": log, + "alog": alog, + "exception": exception, + "aexception": aexception, + } for lvl, name in LEVEL_TO_NAME.items(): meths[name], meths[f"a{name}"] = make_method(lvl) - meths["exception"] = exception - meths["aexception"] = aexception - meths["fatal"] = meths["critical"] - meths["afatal"] = meths["acritical"] - meths["warn"] = meths["warning"] - meths["awarn"] = meths["awarning"] - meths["msg"] = meths["info"] - meths["amsg"] = meths["ainfo"] + for alias, target in [ + ("fatal", "critical"), + ("afatal", "acritical"), + ("warn", "warning"), + ("awarn", "awarning"), + ("msg", "info"), + ("amsg", "ainfo"), + ]: + meths[alias] = meths[target] # Introspection meths["is_enabled_for"] = lambda self, level: level >= min_level diff --git a/src/structlog/contextvars.py b/src/structlog/contextvars.py index 835b6264..299fca0e 100644 --- a/src/structlog/contextvars.py +++ b/src/structlog/contextvars.py @@ -19,6 +19,7 @@ from __future__ import annotations +import asyncio import contextlib import contextvars @@ -44,6 +45,23 @@ contextvars.ContextVar("_ASYNC_CALLING_THREAD") ) +_ASYNC_CALLING_TASK_NAME: contextvars.ContextVar[str | None] = ( + contextvars.ContextVar("_ASYNC_CALLING_TASK_NAME", default=None) +) + + +def _get_current_task_name() -> str | None: + try: + task = asyncio.current_task() + except RuntimeError: + return None + + if task is None: + return None + + return task.get_name() + + # 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. diff --git a/src/structlog/processors.py b/src/structlog/processors.py index 8eefaecc..b56e7693 100644 --- a/src/structlog/processors.py +++ b/src/structlog/processors.py @@ -37,7 +37,11 @@ ) from ._log_levels import NAME_TO_LEVEL, add_log_level from ._utils import get_processname -from .contextvars import _ASYNC_CALLING_THREAD +from .contextvars import ( + _ASYNC_CALLING_TASK_NAME, + _ASYNC_CALLING_THREAD, + _get_current_task_name, +) from .tracebacks import ExceptionDictTransformer from .typing import ( EventDict, @@ -775,6 +779,8 @@ class CallsiteParameter(enum.Enum): PROCESS = "process" #: The name of the process the callsite was executed in. PROCESS_NAME = "process_name" + #: The name of the asyncio task the callsite was executed in. + TASK_NAME = "task_name" def _get_callsite_pathname(module: str, frame: FrameType) -> Any: @@ -829,6 +835,14 @@ def _get_callsite_process_name(module: str, frame: FrameType) -> Any: return get_processname() +def _get_callsite_task_name(module: str, frame: FrameType) -> Any: + task_name = _ASYNC_CALLING_TASK_NAME.get(None) + if task_name is not None: + return task_name + + return _get_current_task_name() + + class CallsiteParameterAdder: """ Adds parameters of the callsite that an event dictionary originated from to @@ -884,6 +898,7 @@ class CallsiteParameterAdder: CallsiteParameter.THREAD_NAME: _get_callsite_thread_name, CallsiteParameter.PROCESS: _get_callsite_process, CallsiteParameter.PROCESS_NAME: _get_callsite_process_name, + CallsiteParameter.TASK_NAME: _get_callsite_task_name, } _record_attribute_map: ClassVar[dict[CallsiteParameter, str]] = { CallsiteParameter.PATHNAME: "pathname", @@ -895,6 +910,7 @@ class CallsiteParameterAdder: CallsiteParameter.THREAD_NAME: "threadName", CallsiteParameter.PROCESS: "process", CallsiteParameter.PROCESS_NAME: "processName", + CallsiteParameter.TASK_NAME: "taskName", } _all_parameters: ClassVar[set[CallsiteParameter]] = set(CallsiteParameter) @@ -944,9 +960,9 @@ def __call__( # then the callsite parameters of the record will not be correct. if record is not None and not from_structlog: for mapping in self._record_mappings: - event_dict[mapping.event_dict_key] = record.__dict__[ + event_dict[mapping.event_dict_key] = record.__dict__.get( mapping.record_attribute - ] + ) return event_dict diff --git a/src/structlog/stdlib.py b/src/structlog/stdlib.py index 2906598d..b9ef027f 100644 --- a/src/structlog/stdlib.py +++ b/src/structlog/stdlib.py @@ -36,7 +36,9 @@ from ._log_levels import LEVEL_TO_NAME, NAME_TO_LEVEL, add_log_level from .contextvars import ( _ASYNC_CALLING_STACK, + _ASYNC_CALLING_TASK_NAME, _ASYNC_CALLING_THREAD, + _get_current_task_name, merge_contextvars, ) from .exceptions import DropEvent @@ -458,6 +460,9 @@ async def _dispatch_to_sync( thread_token = _ASYNC_CALLING_THREAD.set( (threading.get_ident(), threading.current_thread().name) ) + task_name_token = _ASYNC_CALLING_TASK_NAME.set( + _get_current_task_name() + ) scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back.f_back) # type: ignore[union-attr, arg-type, unused-ignore] ctx = contextvars.copy_context() @@ -469,6 +474,7 @@ async def _dispatch_to_sync( finally: _ASYNC_CALLING_STACK.reset(scs_token) _ASYNC_CALLING_THREAD.reset(thread_token) + _ASYNC_CALLING_TASK_NAME.reset(task_name_token) async def adebug(self, event: str, *args: Any, **kw: Any) -> None: """ @@ -666,6 +672,9 @@ async def _dispatch_to_sync( thread_token = _ASYNC_CALLING_THREAD.set( (threading.get_ident(), threading.current_thread().name) ) + task_name_token = _ASYNC_CALLING_TASK_NAME.set( + _get_current_task_name() + ) scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back.f_back) # type: ignore[union-attr, arg-type, unused-ignore] ctx = contextvars.copy_context() @@ -677,6 +686,7 @@ async def _dispatch_to_sync( finally: _ASYNC_CALLING_STACK.reset(scs_token) _ASYNC_CALLING_THREAD.reset(thread_token) + _ASYNC_CALLING_TASK_NAME.reset(task_name_token) async def debug(self, event: str, *args: Any, **kw: Any) -> None: await self._dispatch_to_sync(self.sync_bl.debug, event, args, kw) diff --git a/tests/processors/test_processors.py b/tests/processors/test_processors.py index e29e8027..6f661f04 100644 --- a/tests/processors/test_processors.py +++ b/tests/processors/test_processors.py @@ -5,6 +5,7 @@ from __future__ import annotations +import asyncio import functools import inspect import json @@ -293,6 +294,7 @@ def test_py3_exception_no_traceback(self): class TestCallsiteParameterAdder: parameter_strings = { "pathname", + "task_name", "filename", "module", "func_name", @@ -313,6 +315,107 @@ class TestCallsiteParameterAdder: not in (CallsiteParameter.QUAL_NAME, CallsiteParameter.QUAL_MODULE) } + def test_task_name_structlog(self) -> None: + """ + TASK_NAME is added for structlog-originated events inside an asyncio task. + """ + processor = CallsiteParameterAdder( + parameters={CallsiteParameter.TASK_NAME} + ) + + async def run_test(): + event_dict = {"event": "msg"} + actual = processor(None, None, event_dict) + assert actual["task_name"] is not None + assert actual["task_name"].startswith("Task-") + + asyncio.run(run_test()) + + def test_task_name_structlog_outside_asyncio(self) -> None: + """ + TASK_NAME is None when called synchronously outside of an asyncio task. + """ + processor = CallsiteParameterAdder( + parameters={CallsiteParameter.TASK_NAME} + ) + event_dict: EventDict = {"event": "msg"} + actual = processor(None, None, event_dict) + assert actual["task_name"] is None + + def test_task_name_structlog_in_loop_without_task(self) -> None: + """ + TASK_NAME is None when called inside an event loop but without an active task. + """ + processor = CallsiteParameterAdder( + parameters={CallsiteParameter.TASK_NAME} + ) + loop = asyncio.new_event_loop() + try: + + def callback(): + event_dict: EventDict = {"event": "msg"} + actual = processor(None, None, event_dict) + assert actual["task_name"] is None + + loop.call_soon(callback) + loop.call_soon(loop.stop) + loop.run_forever() + finally: + loop.close() + + def test_task_name_logging_origin_with_task_name(self) -> None: + """ + TASK_NAME is extracted from LogRecord if present. + """ + processor = CallsiteParameterAdder( + parameters={CallsiteParameter.TASK_NAME} + ) + record = logging.LogRecord( + "name", + logging.INFO, + __file__, + 0, + "message", + None, + None, + "func", + ) + record.taskName = "my-custom-task" + event_dict: EventDict = { + "event": "message", + "_record": record, + "_from_structlog": False, + } + actual = processor(None, None, event_dict) + assert "my-custom-task" == actual["task_name"] + + def test_task_name_logging_origin_without_task_name(self) -> None: + """ + TASK_NAME is None from LogRecord if attribute is absent. + """ + processor = CallsiteParameterAdder( + parameters={CallsiteParameter.TASK_NAME} + ) + record = logging.LogRecord( + "name", + logging.INFO, + __file__, + 0, + "message", + None, + None, + "func", + ) + if hasattr(record, "taskName"): + delattr(record, "taskName") + event_dict: EventDict = { + "event": "message", + "_record": record, + "_from_structlog": False, + } + actual = processor(None, None, event_dict) + assert actual["task_name"] is None + def test_all_parameters(self) -> None: """ All callsite parameters are included in ``self.parameter_strings`` and @@ -431,7 +534,7 @@ def __init__(self): @pytest.mark.asyncio async def test_async_native_logger(self) -> None: """ - Callsite thread information for native async invocations is correct. + Callsite thread and task information for native async invocations is correct. """ cf = CapturingLoggerFactory() structlog.configure( @@ -440,6 +543,7 @@ async def test_async_native_logger(self) -> None: parameters=[ CallsiteParameter.THREAD, CallsiteParameter.THREAD_NAME, + CallsiteParameter.TASK_NAME, ] ), ], @@ -459,6 +563,8 @@ async def test_async_native_logger(self) -> None: assert expected_thread == captured["thread"] assert expected_thread_name == captured["thread_name"] + assert captured["task_name"] is not None + assert captured["task_name"].startswith("Task-") def test_additional_ignores(self) -> None: """ @@ -719,6 +825,18 @@ def get_callsite_parameters(cls, offset: int = 1) -> dict[str, object]: """ frame_info = inspect.stack()[1] frame_traceback = inspect.getframeinfo(frame_info[0]) + + def _get_task_name(): + try: + import asyncio + + task = asyncio.current_task() + if task: + return task.get_name() + except RuntimeError: + pass + return None + return { "pathname": frame_traceback.filename, "filename": os.path.basename(frame_traceback.filename), @@ -731,6 +849,7 @@ def get_callsite_parameters(cls, offset: int = 1) -> dict[str, object]: "thread_name": threading.current_thread().name, "process": os.getpid(), "process_name": get_processname(), + "task_name": _get_task_name(), } diff --git a/tests/test_stdlib.py b/tests/test_stdlib.py index 034a5e70..d68044f6 100644 --- a/tests/test_stdlib.py +++ b/tests/test_stdlib.py @@ -31,7 +31,12 @@ from structlog._log_levels import CRITICAL, DEBUG, NAME_TO_LEVEL, WARN from structlog.dev import ConsoleRenderer from structlog.exceptions import DropEvent -from structlog.processors import JSONRenderer, KeyValueRenderer +from structlog.processors import ( + CallsiteParameter, + CallsiteParameterAdder, + JSONRenderer, + KeyValueRenderer, +) from structlog.stdlib import ( AsyncBoundLogger, BoundLogger, @@ -445,6 +450,33 @@ async def test_aexception_exc_info_explicit(self, cl): assert obj is cl.calls[0].kwargs["exc_info"] + @pytest.mark.asyncio + async def test_callsite_task_name(self, capsys): + """ + BoundLogger async methods correctly forward the task name to + CallsiteParameterAdder. + """ + configure( + processors=[ + CallsiteParameterAdder( + parameters=[CallsiteParameter.TASK_NAME] + ), + JSONRenderer(), + ], + logger_factory=PrintLogger, + wrapper_class=BoundLogger, + cache_logger_on_first_use=True, + ) + + logger = get_logger() + + await logger.ainfo("hello bound logger task") + + output = json.loads(capsys.readouterr().out) + assert output["event"] == "hello bound logger task" + assert output["task_name"] is not None + assert output["task_name"].startswith("Task-") + class TestPositionalArgumentsFormatter: def test_formats_tuple(self): @@ -1660,6 +1692,33 @@ async def test_integration(self, capsys): "level": "info", } == json.loads(capsys.readouterr().out) + @pytest.mark.asyncio + async def test_callsite_task_name(self, capsys): + """ + AsyncBoundLogger correctly forwards the task name to + CallsiteParameterAdder. + """ + configure( + processors=[ + CallsiteParameterAdder( + parameters=[CallsiteParameter.TASK_NAME] + ), + JSONRenderer(), + ], + logger_factory=PrintLogger, + wrapper_class=AsyncBoundLogger, + cache_logger_on_first_use=True, + ) + + logger = get_logger() + + await logger.info("hello async bound logger task") + + output = json.loads(capsys.readouterr().out) + assert output["event"] == "hello async bound logger task" + assert output["task_name"] is not None + assert output["task_name"].startswith("Task-") + @pytest.mark.parametrize("log_level", [None, 45]) def test_recreate_defaults(log_level):