Coverage for /usr/local/lib/python3.12/site-packages/prefect/_internal/_logging.py: 70%
18 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1import logging 1a
2import sys 1a
4from typing_extensions import Self 1a
6if sys.version_info < (3, 11): 6 ↛ 8line 6 didn't jump to line 8 because the condition on line 6 was never true1a
8 def getLevelNamesMapping() -> dict[str, int]:
9 return getattr(logging, "_nameToLevel").copy()
10else:
11 getLevelNamesMapping = logging.getLevelNamesMapping # novermin 1a
14class SafeLogger(logging.Logger): 1a
15 """
16 A logger with extensions for safe emission of logs in our concurrency tooling.
17 """
19 def isEnabledFor(self, level: int): 1a
20 # Override `logger.isEnabledFor` to avoid taking a logging lock which can cause
21 # deadlocks during complex concurrency handling
22 from prefect.settings import PREFECT_LOGGING_INTERNAL_LEVEL
24 internal_level = getLevelNamesMapping()[PREFECT_LOGGING_INTERNAL_LEVEL.value()]
26 return level >= internal_level
28 def getChild(self, suffix: str) -> Self: 1a
29 logger = super().getChild(suffix) 1a
30 logger.__class__ = self.__class__ 1a
31 return logger 1a
34# Use `getLogger` to retain `logger.Manager` behavior
35logger = logging.getLogger("prefect._internal") 1a
37# Update the class to inject patched behavior
38logger.__class__ = SafeLogger 1a