diff --git a/sentry_sdk/utils.py b/sentry_sdk/utils.py index 875d28c0d0..4e362cae54 100644 --- a/sentry_sdk/utils.py +++ b/sentry_sdk/utils.py @@ -1,5 +1,4 @@ import base64 -import contextvars import copy import json import linecache @@ -84,29 +83,6 @@ _installed_modules = None -_is_sentry_internal_task = contextvars.ContextVar( - "is_sentry_internal_task", default=False -) - -# These exceptions won't set the span status to error if they occur. Use -# register_control_flow_exception to add to this list -_control_flow_exception_classes: "set[type]" = set() - - -def is_internal_task() -> bool: - return _is_sentry_internal_task.get() - - -@contextmanager -def mark_sentry_task_internal() -> "Generator[None, None, None]": - """Context manager to mark a task as Sentry internal.""" - token = _is_sentry_internal_task.set(True) - try: - yield - finally: - _is_sentry_internal_task.reset(token) - - BASE64_ALPHABET = re.compile(r"^[a-zA-Z0-9/+=]*$") FALSY_ENV_VALUES = frozenset(("false", "f", "n", "no", "off", "0")) @@ -1468,6 +1444,26 @@ def _get_contextvars() -> "Tuple[bool, type]": Please refer to https://docs.sentry.io/platforms/python/contextvars/ for more information. """ +_is_sentry_internal_task = ContextVar("is_sentry_internal_task", default=False) + +# These exceptions won't set the span status to error if they occur. Use +# register_control_flow_exception to add to this list +_control_flow_exception_classes: "set[type]" = set() + + +def is_internal_task() -> bool: + return _is_sentry_internal_task.get() + + +@contextmanager +def mark_sentry_task_internal() -> "Generator[None, None, None]": + """Context manager to mark a task as Sentry internal.""" + token = _is_sentry_internal_task.set(True) + try: + yield + finally: + _is_sentry_internal_task.reset(token) + def qualname_from_function(func: "Callable[..., Any]") -> "Optional[str]": """Return the qualified name of func. Works with regular function, lambda, partial and partialmethod."""