def set_execution_context()

in src/functions_framework/execution_id.py [0:0]

• 25 lines of code
• 3 McCabe index (conditional complexity)

def set_execution_context(request, enable_id_logging=False):
    if enable_id_logging:
        stdout_redirect = contextlib.redirect_stdout(
            LoggingHandlerAddExecutionId(sys.stdout)
        )
        stderr_redirect = contextlib.redirect_stderr(
            LoggingHandlerAddExecutionId(sys.stderr)
        )
    else:
        stdout_redirect = contextlib.nullcontext()
        stderr_redirect = contextlib.nullcontext()

    def decorator(view_function):
        @functools.wraps(view_function)
        def wrapper(*args, **kwargs):
            trace_context = re.match(
                _TRACE_CONTEXT_REGEX_PATTERN,
                request.headers.get(TRACE_CONTEXT_REQUEST_HEADER, ""),
            )
            execution_id = request.headers.get(EXECUTION_ID_REQUEST_HEADER)
            span_id = trace_context.group("span_id") if trace_context else None
            _set_current_context(ExecutionContext(execution_id, span_id))

            with stderr_redirect, stdout_redirect:
                return view_function(*args, **kwargs)

        return wrapper

    return decorator