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

Code before:
    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
        # we want the main variables (I, and R) to stand out visually from the
        # the rest of the variables, so breaking PEP8 is worth it ;)
        R = I = T = Rstr = retval = state = None
        task_request = None
        time_start = monotonic()
        try:
            try:
                kwargs.items
            except AttributeError:
                raise InvalidTaskError(
                    'Task keyword arguments is not a mapping')

            task_request = Context(request or {}, args=args,
                                   called_directly=False, kwargs=kwargs)

Review comment (suggestion):
Again, I think you need to use `Dict` here:

```suggestion
            kwargs: Dict[str, Any],
            request: Optional[Dict[str, Any]] = None) -> trace_ok_t:
```

Code after:
    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
        # we want the main variables (I, and R) to stand out visually from the
        # the rest of the variables, so breaking PEP8 is worth it ;)
        R = I = T = Rstr = retval = state = None
        task_request = None
        time_start = monotonic()
        try:
            try:
                kwargs.items
            except AttributeError:
                raise InvalidTaskError(
                    'Task keyword arguments is not a mapping')

            task_request = Context(request or {}, args=args,
                                   called_directly=False, kwargs=kwargs)