Repository: celery/celery
File: celery/app/trace.py (Python)

Code before:
        task_on_success = task.on_success
    if task_has_custom(task, 'after_return'):
        task_after_return = task.after_return

    pid = os.getpid()

    request_stack = task.request_stack
    push_request = request_stack.push
    pop_request = request_stack.pop
    push_task = _task_stack.push
    pop_task = _task_stack.pop
    _does_info = logger.isEnabledFor(logging.INFO)
    resultrepr_maxsize = task.resultrepr_maxsize

    prerun_receivers = signals.task_prerun.receivers
    postrun_receivers = signals.task_postrun.receivers
    success_receivers = signals.task_success.receivers

    from celery import canvas
    signature = canvas.maybe_signature  # maybe_ does not clone if already

    def on_error(
            request: celery.app.task.Context,
            exc: Union[Exception, Type[Exception]],
            state: str = FAILURE,
            call_errbacks: bool = True) -> tuple[Info, Any, Any, Any]:
        """Handle any errors raised by a `Task`'s execution."""
        if propagate:
            raise
        I = Info(state, exc)
        R = I.handle_error_state(
            task, request, eager=eager, call_errbacks=call_errbacks,
        )
        return I, R, I.state, I.retval

    def trace_task(
            uuid: str,
            args: Sequence[Any],
            kwargs: dict[str, Any],
            request: Optional[dict[str, Any]] = None) -> trace_ok_t:
        """Execute and trace a `Task`."""

        # R      - is the possibly prepared return value.
        # I      - is the Info object.
        # T      - runtime
        # Rstr   - textual representation of return value
        # retval - is the always unmodified return value.
        # state  - is the resulting task state.

        # This function is very long because we've unrolled all the calls
        # for performance reasons, and because the function is so long

Review comment (suggestion):
I think you'll want to use `Tuple` from `typing` here to support Python < 3.9 (I think it was 3.9 which added that...)

```suggestion
            call_errbacks: bool = True) -> Tuple[Info, Any, Any, Any]:
```

Code after:
        task_on_success = task.on_success
    if task_has_custom(task, 'after_return'):
        task_after_return = task.after_return

    pid = os.getpid()

    request_stack = task.request_stack
    push_request = request_stack.push
    pop_request = request_stack.pop
    push_task = _task_stack.push
    pop_task = _task_stack.pop
    _does_info = logger.isEnabledFor(logging.INFO)
    resultrepr_maxsize = task.resultrepr_maxsize

    prerun_receivers = signals.task_prerun.receivers
    postrun_receivers = signals.task_postrun.receivers
    success_receivers = signals.task_success.receivers

    from celery import canvas
    signature = canvas.maybe_signature  # maybe_ does not clone if already

    def on_error(
            request: celery.app.task.Context,
            exc: Union[Exception, Type[Exception]],
            state: str = FAILURE,
            call_errbacks: bool = True) -> Tuple[Info, Any, Any, Any]:
        """Handle any errors raised by a `Task`'s execution."""
        if propagate:
            raise
        I = Info(state, exc)
        R = I.handle_error_state(
            task, request, eager=eager, call_errbacks=call_errbacks,
        )
        return I, R, I.state, I.retval

    def trace_task(
            uuid: str,
            args: Sequence[Any],
            kwargs: Dict[str, Any],
            request: Optional[Dict[str, Any]] = None) -> trace_ok_t:
        """Execute and trace a `Task`."""

        # R      - is the possibly prepared return value.
        # I      - is the Info object.
        # T      - runtime
        # Rstr   - textual representation of return value
        # retval - is the always unmodified return value.
        # state  - is the resulting task state.

        # This function is very long because we've unrolled all the calls
        # for performance reasons, and because the function is so long